codex-app-server/cloud/tests/test_v2cov_marker_pure.py

"""Coverage batch v2 (E2): cloud/marker_tools.py pure helpers --
timezone fallback, first-run parsing, int coercion, spawn-default
parsing, heartbeat/progress formatting, todo/task lines.
Safety buffer on top of E1; all direct calls.
"""

import sys
import time
import unittest
from pathlib import Path
from unittest import mock

REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO_ROOT))

from cloud.marker_tools import (  # noqa: E402
    _as_int,
    _heartbeat_age_text,
    _model_from_match,
    _progress_lines,
    _shanghai_tz,
    format_snapshot,
    format_task,
    format_todo,
    parse_first_run_at,
    parse_spawn_defaults,
    row_get,
)


class FakeMatch:
    def __init__(self, token):
        self._token = token

    def group(self, n):
        return self._token

    def end(self, n):
        return len(self._token)


class MarkerPureTestCase(unittest.TestCase):
    def test_shanghai_tz_zoneinfo_raises(self):
        with mock.patch("cloud.marker_tools.ZoneInfo",
                        side_effect=Exception("no tz")):
            tz = _shanghai_tz()
        self.assertEqual(tz.utcoffset(None).total_seconds(), 8 * 3600)

    def test_shanghai_tz_no_zoneinfo(self):
        with mock.patch("cloud.marker_tools.ZoneInfo", None):
            tz = _shanghai_tz()
        self.assertEqual(tz.utcoffset(None).total_seconds(), 8 * 3600)

    def test_first_run_blank(self):
        self.assertIsNone(parse_first_run_at("   "))

    def test_first_run_negative(self):
        self.assertIsNone(parse_first_run_at(-5))

    def test_first_run_zero(self):
        self.assertIsNone(parse_first_run_at(0))

    def test_first_run_valid(self):
        self.assertEqual(parse_first_run_at(1700000000), 1700000000.0)
        self.assertIsNotNone(parse_first_run_at("2026-01-01T00:00:00Z"))

    def test_row_get_fallback(self):
        self.assertIsNone(row_get(None, "x"))
        self.assertIsNone(row_get({}, "missing"))

    def test_as_int_bool_and_float(self):
        self.assertEqual(_as_int(True, 7), 7)
        self.assertEqual(_as_int(3.5, 7), 7)
        self.assertEqual(_as_int(3.0, 7), 3)
        self.assertEqual(_as_int(4, 7), 4)

    def test_model_from_match_empty_token(self):
        self.assertEqual(_model_from_match("x", FakeMatch("."), 1),
                         (None, None))

    def test_model_from_match_effort_word(self):
        self.assertEqual(_model_from_match("x", FakeMatch("max"), 1),
                         (None, "max"))

    def test_spawn_defaults_empty(self):
        self.assertEqual(parse_spawn_defaults(""), {})
        self.assertEqual(parse_spawn_defaults(None), {})

    def test_spawn_defaults_zh(self):
        out = parse_spawn_defaults("子任务用 gpt-6-luna max")
        self.assertEqual(out.get("model"), "gpt-6-luna")

    def test_heartbeat_none(self):
        self.assertEqual(_heartbeat_age_text({}), "")

    def test_heartbeat_bad(self):
        self.assertEqual(
            _heartbeat_age_text({"last_heartbeat": "bad"}), "")

    def test_heartbeat_future(self):
        now = time.time()
        self.assertEqual(
            _heartbeat_age_text({"last_heartbeat": now + 100}, now=now),
            "0 秒前")

    def test_heartbeat_seconds(self):
        now = time.time()
        self.assertEqual(
            _heartbeat_age_text({"last_heartbeat": now - 30}, now=now),
            "30 秒前")

    def test_progress_lines_full(self):
        now = time.time()
        lines = _progress_lines(
            {"executor": "docker", "actual_model": "m1",
             "worktree_host": "h1", "last_heartbeat": now - 5,
             "turn_id": "t9"}, now=now)
        self.assertEqual(len(lines), 5)

    def test_snapshot_terminal_empty(self):
        text = format_snapshot(
            {"id": "t1", "status": "completed", "result_text": "",
             "error_text": ""}, 100)
        self.assertIn("结果：（空）", text)

    def test_snapshot_timed_out(self):
        text = format_snapshot(
            {"id": "t1", "status": "running", "result_text": "",
             "error_text": ""}, 100, timed_out=True)
        self.assertIn("等待超时", text)

    def test_snapshot_cancelled(self):
        text = format_snapshot(
            {"id": "t1", "status": "running", "result_text": "",
             "error_text": ""}, 100, cancelled=True)
        self.assertIn("被取消", text)

    def test_todo_bad_due_and_task(self):
        line = format_todo({"status": "open", "title": "t", "id": "d1",
                            "priority": "high", "due_at": "bad",
                            "task_id": "t1"})
        self.assertIn("task_id=t1", line)
        self.assertNotIn("due=", line)

    def test_task_bad_created(self):
        line = format_task({"status": "completed", "id": "t1",
                            "created_by": "u1", "prompt": "do it",
                            "created_at": "bad"})
        self.assertIn("do it", line)
        self.assertNotIn("created=", line)