host
12e72dab9b
Baseline merge head e5ff26a: 28144 stmts, 2807 missing, exact 90.0263%, margin +7. After thickening (full run): 2350 missing, exact 91.6501%, margin +464. Full suite 3516 items: 3509 pass, 1 failure (pre-existing November date failure, untouched), 6 skipped. No production code changed.
565 lines
22 KiB
Python
565 lines
22 KiB
Python
"""Coverage batch v4-A: evidence modules, handler close/502 branches,
|
|
server defensive branches, transplant v05 newline branch.
|
|
|
|
Targets are the baseline-missing lines of the v4 merge head, verified
|
|
against the subset-coverage intersection (see the v4 summary). Stubbing
|
|
discipline: the method under test is never stubbed. Collaborators that
|
|
are stubbed are: CostStore / _validate_sub2api_key /
|
|
_list_sub2api_models (handler_identity namespace -- the v2 batch patched
|
|
cloud.server by mistake after the split, which is why these lines stayed
|
|
uncovered), _check_project / _project_exists predicates for the two
|
|
project 404 branches (unreachable through the real check, which 404s
|
|
earlier when the project row is absent), parse_paths_field for the
|
|
upload count-mismatch branch (the real parser raises first, so the
|
|
handler's own duplicate check is defensive), os.environ.get for the
|
|
_default_executor None branch (a real environ value can never be
|
|
None), and the _connect/_init/_ensure helpers for the init_db
|
|
Postgres branch (no live PG in this environment).
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import cloud.server as server_mod
|
|
from cloud import evidence_binding
|
|
from cloud import gate_verification
|
|
from cloud import ledger_onboarding
|
|
from cloud import ledger_outbox
|
|
from cloud.evidence_binding import (
|
|
EvidenceNonceStore,
|
|
bind_evidence,
|
|
binding_presence,
|
|
validate_bound_evidence,
|
|
)
|
|
from cloud.gate_verification import verify_gate_evidence
|
|
from cloud.ledger_onboarding import _table_columns, verify_ledger_ready
|
|
from cloud.ledger_outbox import (
|
|
append_failure,
|
|
outbox_path_for,
|
|
read_failures,
|
|
replay_failures,
|
|
)
|
|
from cloud.server import CloudControlPlaneHandler, init_db
|
|
|
|
|
|
def _handler(db_path):
|
|
h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
|
|
h.server = SimpleNamespace(db_path=db_path)
|
|
sent = []
|
|
h._send_json = lambda status, obj, extra_headers=None: sent.append(
|
|
(status, obj))
|
|
h.sent = sent
|
|
return h
|
|
|
|
|
|
def _conn(db_path):
|
|
c = sqlite3.connect(db_path)
|
|
c.row_factory = sqlite3.Row
|
|
return c
|
|
|
|
|
|
class _Base(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "test.db")
|
|
init_db(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _h(self, body=None, err=None):
|
|
h = _handler(self.db_path)
|
|
h._read_json = lambda: (body, err)
|
|
return h
|
|
|
|
def _project(self, pid="p1", owner="u1"):
|
|
from cloud.server import _auth_store
|
|
c = _conn(self.db_path)
|
|
c.execute(
|
|
"INSERT INTO projects(id, name, owner_id, created_at,"
|
|
" repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
|
|
(pid, "P", owner, time.time(), None))
|
|
c.commit()
|
|
c.close()
|
|
with _auth_store(self.db_path) as store:
|
|
store.create_project(pid, owner)
|
|
|
|
def _task(self, tid="t1", pid="p1", status="completed"):
|
|
c = _conn(self.db_path)
|
|
c.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
|
" created_by, created_at, updated_at)"
|
|
" VALUES(?, ?, ?, ?, ?, ?, ?)",
|
|
(tid, pid, "do work", status, "u1", time.time(),
|
|
time.time()))
|
|
c.commit()
|
|
c.close()
|
|
|
|
def _last(self, h):
|
|
self.assertTrue(h.sent, "handler sent no response")
|
|
return h.sent[-1]
|
|
|
|
|
|
class EvidenceBindingGapTestCase(unittest.TestCase):
|
|
def test_bind_non_dict_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
bind_evidence(["not", "a", "dict"])
|
|
|
|
def test_validate_non_dict(self):
|
|
self.assertEqual(validate_bound_evidence("x"),
|
|
(False, "evidence_binding_missing"))
|
|
|
|
def test_validate_both_missing(self):
|
|
self.assertEqual(validate_bound_evidence({}),
|
|
(False, "evidence_binding_missing"))
|
|
|
|
def test_validate_bool_ts(self):
|
|
self.assertEqual(
|
|
validate_bound_evidence(
|
|
{"evidence_ts": True, "evidence_nonce": "n1"}),
|
|
(False, "evidence_ts_not_a_number"))
|
|
|
|
def test_validate_blank_nonce(self):
|
|
self.assertEqual(
|
|
validate_bound_evidence(
|
|
{"evidence_ts": 1000.0, "evidence_nonce": " "}),
|
|
(False, "evidence_nonce_empty"))
|
|
|
|
def test_binding_presence_non_dict(self):
|
|
self.assertEqual(binding_presence(None), "absent")
|
|
|
|
def test_has_exception_returns_false(self):
|
|
class _RaisingConn:
|
|
def execute(self, *args):
|
|
raise RuntimeError("db gone")
|
|
store = EvidenceNonceStore(_RaisingConn())
|
|
self.assertFalse(store.has("n1"))
|
|
|
|
|
|
class GateVerificationGapTestCase(unittest.TestCase):
|
|
def _live(self, ci_checks):
|
|
return {"ci_checks": ci_checks, "branch_protection": "protected"}
|
|
|
|
def test_reader_non_dict_result(self):
|
|
out = verify_gate_evidence(
|
|
{}, reader=lambda repo, ref: ["not", "a", "dict"],
|
|
repo="r", ref="s")
|
|
self.assertFalse(out["checked"])
|
|
self.assertFalse(out["verified"])
|
|
self.assertIn("non-dict", out["reason"])
|
|
|
|
def test_live_checks_non_dict_treated_empty(self):
|
|
out = verify_gate_evidence(
|
|
{}, reader=lambda repo, ref: self._live("nope"),
|
|
repo="r", ref="s")
|
|
self.assertTrue(out["checked"])
|
|
self.assertTrue(out["verified"])
|
|
|
|
def test_ci_checks_non_dict_item_skipped(self):
|
|
out = verify_gate_evidence(
|
|
{"ci_checks": ["oops", {"name": "ci", "status": "success"}]},
|
|
reader=lambda repo, ref: self._live({"ci": "success"}),
|
|
repo="r", ref="s")
|
|
self.assertTrue(out["checked"])
|
|
self.assertTrue(out["verified"])
|
|
self.assertEqual(out["mismatches"], [])
|
|
|
|
def test_ci_checks_nameless_item_skipped(self):
|
|
out = verify_gate_evidence(
|
|
{"ci_checks": [{"status": "success"}]},
|
|
reader=lambda repo, ref: self._live({}),
|
|
repo="r", ref="s")
|
|
self.assertTrue(out["checked"])
|
|
self.assertTrue(out["verified"])
|
|
|
|
|
|
class _FakeConn:
|
|
def __init__(self, rows=None, exc=None):
|
|
self._rows = rows
|
|
self._exc = exc
|
|
|
|
def execute(self, sql, params=()):
|
|
if self._exc is not None:
|
|
raise self._exc
|
|
result = mock.Mock()
|
|
result.fetchall.return_value = self._rows
|
|
return result
|
|
|
|
|
|
class LedgerOnboardingGapTestCase(unittest.TestCase):
|
|
def test_table_columns_exception_returns_none(self):
|
|
self.assertIsNone(_table_columns(
|
|
_FakeConn(exc=RuntimeError("pg dialect")), "task_attempts"))
|
|
|
|
def test_table_columns_malformed_rows_return_none(self):
|
|
# Row tuples too short for row[1]: IndexError -> None.
|
|
self.assertIsNone(_table_columns(_FakeConn(rows=[(1,)]),
|
|
"task_attempts"))
|
|
|
|
def _sqlite(self, ledger_cols, marker_cols):
|
|
conn = sqlite3.connect(":memory:")
|
|
if ledger_cols is not None:
|
|
conn.execute("CREATE TABLE task_attempts(%s)"
|
|
% ", ".join("%s TEXT" % c for c in ledger_cols))
|
|
if marker_cols is not None:
|
|
conn.execute("CREATE TABLE task_attempt_origin_markers(%s)"
|
|
% ", ".join("%s TEXT" % c for c in marker_cols))
|
|
return conn
|
|
|
|
def test_verify_ledger_missing_column(self):
|
|
conn = self._sqlite(
|
|
["task_id", "attempt_no", "outcome", "is_defect",
|
|
"defect_category", "created_at"],
|
|
["task_id", "origin", "queued_at"])
|
|
try:
|
|
out = verify_ledger_ready(conn)
|
|
finally:
|
|
conn.close()
|
|
self.assertFalse(out["tables_ready"])
|
|
self.assertEqual(
|
|
out["missing"],
|
|
["task_attempts missing columns: corrects_attempt_no"])
|
|
|
|
def test_verify_marker_missing_column(self):
|
|
conn = self._sqlite(
|
|
["task_id", "attempt_no", "outcome", "is_defect",
|
|
"defect_category", "created_at", "corrects_attempt_no"],
|
|
["task_id", "origin"])
|
|
try:
|
|
out = verify_ledger_ready(conn)
|
|
finally:
|
|
conn.close()
|
|
self.assertFalse(out["tables_ready"])
|
|
self.assertEqual(
|
|
out["missing"],
|
|
["task_attempt_origin_markers missing columns: queued_at"])
|
|
|
|
|
|
class LedgerOutboxGapTestCase(unittest.TestCase):
|
|
def test_outbox_path_explicit_env_wins(self):
|
|
self.assertEqual(
|
|
outbox_path_for("/x/db.sqlite",
|
|
env={"CLOUD_LEDGER_FAILURE_OUTBOX": "/o/f.jsonl"}),
|
|
"/o/f.jsonl")
|
|
|
|
def test_outbox_path_memory_and_empty_none(self):
|
|
self.assertIsNone(outbox_path_for(":memory:", env={}))
|
|
self.assertIsNone(outbox_path_for("", env={}))
|
|
|
|
def test_append_failure_falsy_path(self):
|
|
self.assertFalse(append_failure("", "t1", "failure", "boom"))
|
|
self.assertFalse(append_failure(None, "t1", "failure", "boom"))
|
|
|
|
def test_read_failures_missing_file(self):
|
|
self.assertEqual(read_failures("/definitely/not/here.jsonl"),
|
|
([], 0))
|
|
|
|
def test_read_failures_bad_lines_counted(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "out.jsonl")
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write('{"not json\n')
|
|
fh.write('\n')
|
|
fh.write(json.dumps({"task_id": "t1",
|
|
"outcome": "weird"}) + "\n")
|
|
fh.write(json.dumps([1, 2]) + "\n")
|
|
fh.write(json.dumps({"task_id": "t2",
|
|
"outcome": "success"}) + "\n")
|
|
records, bad = read_failures(path)
|
|
self.assertEqual([r["task_id"] for r in records], ["t2"])
|
|
self.assertEqual(bad, 3)
|
|
|
|
def test_replay_failed_record_kept_and_rewritten(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "out.jsonl")
|
|
self.assertTrue(append_failure(path, "t9", "failure", "db"))
|
|
def _boom(conn, task_id, outcome, is_defect=False):
|
|
raise RuntimeError("still down")
|
|
stats = replay_failures(path, None, record_fn=_boom)
|
|
self.assertEqual(stats["replayed"], 0)
|
|
self.assertEqual(stats["failed"], 1)
|
|
self.assertEqual(stats["remaining"], 1)
|
|
self.assertEqual(stats["bad_lines"], 0)
|
|
records, bad = read_failures(path)
|
|
self.assertEqual([r["task_id"] for r in records], ["t9"])
|
|
self.assertEqual(bad, 0)
|
|
|
|
def test_replay_directory_path_write_failure_swallowed(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
stats = replay_failures(d, None,
|
|
record_fn=lambda *a, **k: None)
|
|
self.assertEqual(stats, {"replayed": 0, "failed": 0,
|
|
"remaining": 0, "bad_lines": 0})
|
|
|
|
def test_replay_bad_lines_not_rewritten(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "out.jsonl")
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write('{"broken\n')
|
|
fh.write(json.dumps({"task_id": "t3",
|
|
"outcome": "success"}) + "\n")
|
|
seen = []
|
|
def _ok(conn, task_id, outcome, is_defect=False):
|
|
seen.append(task_id)
|
|
stats = replay_failures(path, None, record_fn=_ok)
|
|
self.assertEqual(stats["replayed"], 1)
|
|
self.assertEqual(stats["bad_lines"], 1)
|
|
self.assertEqual(stats["remaining"], 1)
|
|
self.assertEqual(seen, ["t3"])
|
|
with open(path, encoding="utf-8") as fh:
|
|
self.assertIn('{"broken', fh.read())
|
|
|
|
|
|
class HandlerGapTestCase(_Base):
|
|
def test_project_cost_close_failure_swallowed(self):
|
|
class _Store:
|
|
def __init__(self, db_path):
|
|
self._conn = mock.Mock()
|
|
self._conn.close.side_effect = RuntimeError("down")
|
|
|
|
def dual_column_report_by_project(self, project_id):
|
|
return {}
|
|
h = self._h()
|
|
with mock.patch("cloud.handler_identity.CostStore", _Store):
|
|
h._project_cost("p1", {})
|
|
self.assertEqual(self._last(h)[0], 200)
|
|
|
|
def test_task_cost_close_failure_swallowed(self):
|
|
self._project()
|
|
self._task()
|
|
class _Store:
|
|
def __init__(self, db_path):
|
|
self._conn = mock.Mock()
|
|
self._conn.close.side_effect = RuntimeError("down")
|
|
|
|
def dual_column_report_by_task(self, task_id):
|
|
return {}
|
|
h = self._h()
|
|
with mock.patch("cloud.handler_identity.CostStore", _Store):
|
|
h._task_cost("u1", "t1")
|
|
self.assertEqual(self._last(h)[0], 200)
|
|
|
|
def test_relay_models_sub2api_generic_error_502(self):
|
|
h = self._h()
|
|
with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
|
|
mock.patch("cloud.handler_identity._validate_sub2api_key",
|
|
side_effect=RuntimeError("net down")):
|
|
h._relay_models("u1")
|
|
self.assertEqual(self._last(h)[0], 502)
|
|
|
|
def test_relay_models_sub2api_list_error_502(self):
|
|
h = self._h()
|
|
with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
|
|
mock.patch("cloud.handler_identity._validate_sub2api_key",
|
|
return_value=("ok", "key", "sub", "kid")), \
|
|
mock.patch("cloud.handler_identity._list_sub2api_models",
|
|
side_effect=RuntimeError("gw down")):
|
|
h._relay_models("u1")
|
|
self.assertEqual(self._last(h)[0], 502)
|
|
|
|
def test_models_overview_close_failure_swallowed(self):
|
|
self._project()
|
|
conn = mock.Mock()
|
|
conn.execute.return_value.fetchall.return_value = []
|
|
conn.close.side_effect = RuntimeError("down")
|
|
|
|
class _Store:
|
|
def __init__(self, db_path):
|
|
self._conn = conn
|
|
h = self._h()
|
|
with mock.patch("cloud.handler_identity.CostStore", _Store):
|
|
h._models_overview("u1")
|
|
self.assertEqual(self._last(h)[0], 200)
|
|
|
|
def test_create_environment_project_row_gone_404(self):
|
|
h = self._h(body={"project_id": "pX", "name": "env1"})
|
|
h._check_project = lambda *args: None
|
|
h._create_environment("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 404)
|
|
self.assertEqual(payload["error"], "project not found")
|
|
|
|
def test_update_project_row_gone_404(self):
|
|
h = self._h(body={"repo_url": ""})
|
|
h._check_project = lambda *args: None
|
|
h._update_project("u1", "pX")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 404)
|
|
self.assertEqual(payload["error"], "project not found")
|
|
|
|
|
|
class ServerGapTestCase(_Base):
|
|
def test_read_json_empty_body(self):
|
|
h = _handler(self.db_path)
|
|
h._read_body = lambda: b""
|
|
self.assertEqual(h._read_json(), ({}, None))
|
|
|
|
def test_default_executor_none_env_value(self):
|
|
real_get = os.environ.get
|
|
|
|
def _fake_get(key, default=None):
|
|
if key == "CLOUD_DEFAULT_EXECUTOR":
|
|
return None
|
|
return real_get(key, default)
|
|
with mock.patch.object(os.environ, "get", side_effect=_fake_get):
|
|
self.assertEqual(server_mod._default_executor(), "docker")
|
|
|
|
def _backfill_conn(self, mode, rollback_raises=False):
|
|
class _Fake:
|
|
def __init__(self):
|
|
self.rollback_calls = 0
|
|
self.rollback_raises = rollback_raises
|
|
self.mode = mode
|
|
|
|
def execute(self, sql, params=()):
|
|
text = sql.strip()
|
|
if text.startswith("ROLLBACK TO"):
|
|
raise RuntimeError("rollback-to down")
|
|
if text.startswith("SAVEPOINT"):
|
|
if self.mode == "savepoint":
|
|
raise RuntimeError("savepoint down")
|
|
return mock.Mock()
|
|
if "INSERT OR IGNORE INTO user_quotas" in text:
|
|
raise RuntimeError("insert down")
|
|
return mock.Mock()
|
|
|
|
def rollback(self):
|
|
self.rollback_calls += 1
|
|
if self.rollback_raises:
|
|
raise RuntimeError("rollback down")
|
|
return _Fake()
|
|
|
|
def test_backfill_savepoint_failure_rollback_ok(self):
|
|
conn = self._backfill_conn("savepoint")
|
|
self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
|
|
self.assertEqual(conn.rollback_calls, 1)
|
|
|
|
def test_backfill_savepoint_failure_rollback_raises(self):
|
|
conn = self._backfill_conn("savepoint", rollback_raises=True)
|
|
self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
|
|
self.assertEqual(conn.rollback_calls, 1)
|
|
|
|
def test_backfill_insert_failure_rollback_to_failure(self):
|
|
conn = self._backfill_conn("insert")
|
|
self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
|
|
self.assertEqual(conn.rollback_calls, 1)
|
|
|
|
def test_backfill_insert_failure_rollback_raises(self):
|
|
conn = self._backfill_conn("insert", rollback_raises=True)
|
|
self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
|
|
self.assertEqual(conn.rollback_calls, 1)
|
|
|
|
def test_init_db_postgres_branch(self):
|
|
conn = mock.Mock()
|
|
with mock.patch.dict(os.environ,
|
|
{"CLOUD_DB_BACKEND": "postgres"}), \
|
|
mock.patch("cloud.server._connect",
|
|
return_value=conn), \
|
|
mock.patch("cloud.server._init_postgres_schema"), \
|
|
mock.patch("cloud.server.migrate_tasks_table"), \
|
|
mock.patch("cloud.server.migrate_projects_table"), \
|
|
mock.patch("cloud.server._ensure_followups_table"), \
|
|
mock.patch("cloud.server._ensure_continue_messages_table"), \
|
|
mock.patch("cloud.server._ensure_environments_table"), \
|
|
mock.patch("cloud.server._ensure_quotas_table"), \
|
|
mock.patch("cloud.server._backfill_default_quotas"), \
|
|
mock.patch("cloud.server._ensure_gateway_keys"), \
|
|
mock.patch("cloud.server._ensure_uploads_tables"), \
|
|
mock.patch("cloud.server._ensure_task_checkpoints_table"), \
|
|
mock.patch("cloud.server._ensure_task_steps_table"):
|
|
init_db("/nonexistent/pg.db")
|
|
conn.commit.assert_called_once_with()
|
|
|
|
def _multipart(self, boundary, fields, files):
|
|
out = b""
|
|
for name, value in fields.items():
|
|
out += (("--%s\r\nContent-Disposition: form-data; "
|
|
"name=\"%s\"\r\n\r\n%s\r\n")
|
|
% (boundary, name, value)).encode("utf-8")
|
|
for part_name, filename, data in files:
|
|
out += (("--%s\r\nContent-Disposition: form-data; "
|
|
"name=\"%s\"; filename=\"%s\"\r\n\r\n")
|
|
% (boundary, part_name, filename)).encode("utf-8")
|
|
out += data + b"\r\n"
|
|
out += ("--%s--\r\n" % boundary).encode("utf-8")
|
|
return out
|
|
|
|
def test_create_upload_paths_files_mismatch(self):
|
|
boundary = "TESTBOUNDARY"
|
|
raw = self._multipart(
|
|
boundary, {"paths": '["a.txt", "b.txt"]'},
|
|
[("file0", "a.txt", b"hello"), ("file1", "b.txt", b"world")])
|
|
h = _handler(self.db_path)
|
|
h.headers = {"Content-Type":
|
|
"multipart/form-data; boundary=" + boundary,
|
|
"Content-Length": str(len(raw))}
|
|
h._read_body = lambda: raw
|
|
with mock.patch("cloud.server.parse_paths_field",
|
|
return_value=["only-one.txt"]):
|
|
h._create_upload("u1")
|
|
code, payload = h.sent[-1]
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("不一致", payload["error"])
|
|
|
|
def test_release_task_bound_bad_ts_400(self):
|
|
self._task()
|
|
body = {
|
|
"files_changed": ["apps/web/a.py"],
|
|
"owned_dir": "apps/web",
|
|
"branch_protection": "protected",
|
|
"ci_status": "success",
|
|
"ci_checks": [{"name": "ci", "status": "success"}],
|
|
"required_checks": ["ci"],
|
|
"contract_version": "v2",
|
|
"contract_frozen": True,
|
|
"coordinator_approved": True,
|
|
"unresolved_review_comments": 0,
|
|
"evidence_ts": "not-a-number",
|
|
"evidence_nonce": "n1",
|
|
}
|
|
h = self._h(body=body)
|
|
with mock.patch.dict(os.environ,
|
|
{"CODEX_GLOBAL_ADMINS": "admin1"}):
|
|
h._release_task("admin1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertEqual(payload["error"], "evidence_ts_not_a_number")
|
|
|
|
|
|
class TransplantGapTestCase(unittest.TestCase):
|
|
def test_transform_v05_notify_with_trailing_newline(self):
|
|
from importlib import util as _util
|
|
script = (REPO_ROOT / "cloud" / "skills" / "software-engineering"
|
|
/ "tools" / "check_transplant.py")
|
|
spec = _util.spec_from_file_location("check_transplant_v4a",
|
|
str(script))
|
|
module = _util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
lines = [
|
|
"前言。" + module.AUDIT_WARNING,
|
|
module.TIGHTEN3_NOTIFY,
|
|
"filler line after notify",
|
|
module.QA_HEADER,
|
|
"qa body",
|
|
]
|
|
out_lines, stopped = module.transform_main(lines, "v05")
|
|
text = "\n".join(out_lines)
|
|
self.assertNotIn(module.TIGHTEN3_NOTIFY, text)
|
|
self.assertTrue(any("QA 全链路" in s for s in stopped))
|
|
self.assertTrue(any("收紧三" in s for s in stopped))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|