# coding=utf-8
from datetime import datetime
from unittest.mock import patch

import pytest

from spark_executor.core import connection_store
from spark_executor.core.job_store import JobStore
from spark_executor.models import Connection, Job
from spark_executor.tools import logs
from spark_executor.tools import connections


def _fresh_stores():
    store = connection_store.ConnectionStore()
    connection_store.store = store
    connections.store = store
    logs.conn_store = store
    logs.store = JobStore()


def _seed(job_id="abc", app_id="application_1"):
    _fresh_stores()
    logs.conn_store.save(Connection(name="prod", master="yarn", yarn_rm_url="http://rm:8088"))
    logs.store.put(
        Job(
            job_id=job_id,
            application_id=app_id,
            script_path="/tmp/j.py",
            queue="default",
            submit_time=datetime(2026, 6, 24),
            connection="prod",
            yarn_rm_url="http://rm:8088",
        )
    )


def test_get_job_logs_tails_to_default_5000():
    _seed()
    big = "x" * 8000 + "\nEND"
    with patch("spark_executor.tools.logs.get_application_logs", return_value=big):
        out = logs.get_job_logs("abc")
    assert out.endswith("END")
    assert len(out) == 5000


def test_get_job_logs_respects_custom_tail_chars():
    _seed()
    with patch(
        "spark_executor.tools.logs.get_application_logs",
        return_value="0123456789",
    ):
        out = logs.get_job_logs("abc", tail_chars=3)
    assert out == "789"


def test_get_job_logs_raises_for_unknown_job():
    _seed()
    with pytest.raises(KeyError):
        logs.get_job_logs("missing")


def test_get_job_logs_raises_when_connection_missing():
    _fresh_stores()
    logs.store.put(
        Job(
            job_id="abc",
            application_id="application_1",
            script_path="/tmp/j.py",
            queue="default",
            submit_time=datetime(2026, 6, 24),
            connection="missing",
        )
    )
    with pytest.raises(KeyError, match="Connection not found"):
        logs.get_job_logs("abc")