"""Historical imports retain source facts and never start model execution.""" import copy import base64 import json import sqlite3 import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from weknora_business_etl import (ADMIN_TARGET_ID, HISTORY_TABLES, JSON_FIELDS, LEDGER, RESOURCE_KINDS, business_id, business_plan, business_sql, safe_business_summary, postgres_plan) USER = "ab043a7d-d10e-58fe-8417-3fd2cbab5f77" AGENT = "72a9a19f-65f3-4996-897b-efca38654b70" KB = "7e6ac485-f766-4d58-ae38-2f0456a022ef" class BusinessTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.artifact = self.root / "seg_1_r1.md" self.artifact.write_text(" Original paragraph O'Brien\n\n", encoding="utf-8") self.db = sqlite3.connect(":memory:") self.addCleanup(self.db.close) shared = dict(organization_id=1, owner_user_id=3, visibility="private", created_at="2026-09-15 01:00:00") flow = dict(id=1, name="Flow", nodes=[{"node_id": "a", "agent_id": 9, "enabled": True, "x": 20, "y": 40}], edges=[], supervisor_agent_id=None, supervisor_mode="off", **shared) map_row = dict(id=2, owner_user_id=3, organization_id=1, topic="Course", kb_ids=[5], kb_category_ids={}, requested_days=1, agent_id=9, version=2, confirmed_version=2, plan={"recommended_days": 1, "rationale": "Original rationale", "insufficiencies": [], "days": [{"day": 1, "title": "Day one", "source_ids": ["E1"], "support": "Original support", "prerequisite": "Prerequisite"}]}, evidence=[{"id": "E1", "text": "Private evidence"}], coverage={"scope": "quick"}, assessment_cancel_requested=0, maps={"1": {"day": 1, "title": "Daily framework", "children": [ {"title": "Node title", "content": "Node content", "source_ids": ["E1"], "hook": "Original hook"}]}}, status="confirmed", error="") campaign = dict(id=3, name="Task", mindmap_text="Original map text", objective="Legacy objective", success_criteria="Legacy criteria", canvas_snapshot={"nodes": flow["nodes"], "edges": []}, canvas_name="Flow", duration_minutes=60, qc_mode="full", previous_campaign_id=None, task_spec=[{"title": "Original paragraph"}], pending_confirms=["Original pending"], notes=[], status="human_review", has_failure=0, finished_at="", **shared) segment = dict(id=4, campaign_id=3, node_id="a", agent_id=9, agent_assignment_mode="manual", generation_history=[], fusion_agent_ids=[], runtime_snapshot={"primary": {"rules": "Private frozen rules"}}, content_tag="Tag", title="Original paragraph", time_range="00:00-00:10", mindmap_node="Source node", product_ref="", status="approved", round=1, artifact_path=str(self.artifact), anchor_card={}, qc_reports=[], failure={}, references=[{"chunk_id": "private-chunk", "text": "Private reference"}], kb_category_ids=[], kb_category_recommendation={}, updated_at="2026-09-15 02:00:00") run = dict(id=5, campaign_id=3, user_id=3, organization_id=1, status="completed", cancel_requested=0, pause_requested=0, checkpoint={"step": "review"}, config_snapshot={"rules": "Original runtime config"}, trace_id="original-trace", runtime_version="v1", error="", created_at="2026-09-15 01:01:00", started_at="2026-09-15 01:01:00", finished_at="2026-09-15 02:00:00") report = dict(id=6, segment_id=4, gate="cross", checks=[{"result": "Original failed check"}], verdict="FAIL", run_id=5, trace_id="original-trace", runtime_snapshot={}, created_at="2026-09-15 01:40:00") event = dict(id=7, run_id=5, campaign_id=3, segment_id=4, node_id="a", trace_id="original-trace", span_id="span", parent_span_id="", actor_agent_id=9, actor_name="Agent", tool_name="knowledge_search", event_type="error", status="failed", payload={"error": "Original model error"}, created_at="2026-09-15 01:30:00") session = dict(id=8, run_id=5, campaign_id=3, segment_id=4, agent_id=9, agent_name="Agent", parent_session_id=None, phase="production", trace_id="original-trace", status="completed", turn=1, max_turns=6, messages=[{"role": "system", "content": "Private historical prompt"}], tool_state={}, result_text="Original model result", result_meta={}, error="", created_at="2026-09-15 01:10:00", updated_at="2026-09-15 01:20:00", finished_at="2026-09-15 01:20:00") queue = dict(id=9, task_key="campaign-run:5", payload={"run_id": 5, "campaign_id": 3}, status="done", attempt=1) rows = {"canvas": [flow], "trainingmap": [map_row], "campaign": [campaign], "segment": [segment], "campaignrun": [run], "qcreport": [report], "runevent": [event], "agentsession": [session], "taskqueueitem": [queue]} for table in (*RESOURCE_KINDS, *HISTORY_TABLES): sample = rows.get(table, [{"id": 0}])[0] columns = ",".join('"' + key + '"' + (" INTEGER" if type(value) is int else " TEXT") for key, value in sample.items()) self.db.execute(f'CREATE TABLE "{table}" ({columns})') for row in rows.get(table, []): self.insert(table, row) self.db.commit() self.db.execute("PRAGMA query_only=ON") self.identities = {"1": ADMIN_TARGET_ID, "3": USER} def insert(self, table, row): columns = ",".join('"' + key + '"' for key in row) values = [json.dumps(value) if key in JSON_FIELDS.get(table, ()) else value for key, value in row.items()] self.db.execute(f'INSERT INTO "{table}" ({columns}) VALUES (' + ",".join("?" for _ in row) + ")", values) def change(self, sql, values=()): self.db.execute("PRAGMA query_only=OFF") self.db.execute(sql, values) self.db.execute("PRAGMA query_only=ON") def plan(self): result = business_plan(self.db, self.identities, {"9": AGENT}, {"5": KB}, self.root) result["source_snapshot_sha256"] = "a" * 64 return result def resource(self, plan, kind): return next(r["target"] for r in plan["resources"] if r["source_kind"] == kind) def test_maps_agents_owners_and_knowledge_without_inference(self): plan = self.plan() flow = self.resource(plan, "canvas") self.assertEqual(flow["owner_id"], USER) self.assertEqual(flow["data"]["nodes"][0]["agent_id"], AGENT) self.assertEqual(flow["data"]["legacy_history"]["nodes"][0]["x"], 20) training = self.resource(plan, "trainingmap") self.assertEqual(training["data"]["knowledge_base_ids"], [KB]) self.assertEqual(training["version"], 2) task = self.resource(plan, "campaign") self.assertEqual(task["data"]["map_id"], "") self.assertEqual(task["data"]["flow_id"], "") self.assertEqual(task["data"]["day"], 0) def test_preserves_native_text_and_source_status(self): plan = self.plan() task = self.resource(plan, "campaign") segment = task["data"]["segments"][0] self.assertEqual(segment["content"], " Original paragraph O'Brien\n\n") self.assertEqual(segment["status"], "approved") self.assertEqual(task["status"], "human_review") self.assertEqual(task["data"]["composed"], "") self.assertIn('"verdict": "FAIL"', segment["review"]) self.assertIn("Original failed check", task["data"]["review"]) def test_daily_framework_preserves_node_content_without_forging_confirmation(self): data = self.resource(self.plan(), "trainingmap")["data"] day = data["days"][0] self.assertEqual(day["nodes"][0]["label"], "Node title") self.assertIn("Node content", day["framework"]) self.assertIn("Original hook", day["framework"]) self.assertFalse(day["confirmed"]) self.assertEqual(data["legacy_history"]["confirmed_version"], 2) self.assertEqual(data["legacy_history"]["plan"]["days"][0]["support"], "Original support") def test_every_resource_blocks_automatic_execution(self): for resource in self.plan()["resources"]: marker = resource["target"]["data"]["legacy_import"] self.assertTrue(marker["read_only"]) self.assertTrue(marker["requires_review"]) self.assertEqual(marker["source_id"], resource["source_id"]) task = self.resource(self.plan(), "campaign") self.assertTrue(task["data"]["legacy_import"]["needs_configuration"]) def test_missing_agent_and_category_scope_are_not_widened(self): self.change("UPDATE trainingmap SET agent_id=NULL,kb_category_ids=?", (json.dumps({"5": [9]}),)) data = self.resource(self.plan(), "trainingmap")["data"] self.assertEqual(data["agent_id"], "") self.assertIn("knowledge_category_scope", data["legacy_import"]["missing_configuration"]) self.assertTrue(data["legacy_import"]["read_only"]) def test_empty_canvas_remains_empty_and_requires_configuration(self): self.change("UPDATE canvas SET nodes='[]'") data = self.resource(self.plan(), "canvas")["data"] self.assertEqual(data["nodes"], []) self.assertIn("flow_nodes", data["legacy_import"]["missing_configuration"]) def test_unmapped_or_deleted_owner_is_not_reassigned_to_admin(self): self.change("UPDATE canvas SET owner_user_id=2") with self.assertRaisesRegex(ValueError, "no durable identity"): self.plan() def test_invalid_identity_organization_or_snapshot_rejected(self): self.change("UPDATE canvas SET organization_id=2") with self.assertRaisesRegex(ValueError, "organization"): self.plan() self.change("UPDATE canvas SET organization_id=1") self.db.execute("PRAGMA query_only=OFF") with self.assertRaisesRegex(ValueError, "read-only"): self.plan() def test_runtime_and_history_go_to_ledger_without_native_jobs_or_events(self): plan = self.plan() sources = {r["source_kind"]: r["source"] for r in plan["history"]} self.assertEqual(sources["agentsession"]["messages"][0]["content"], "Private historical prompt") self.assertEqual(sources["runevent"]["payload"]["error"], "Original model error") self.assertEqual(sources["taskqueueitem"]["status"], "done") sql = business_sql(plan).upper() for table in ("WORKBENCH_JOBS", "WORKBENCH_EVENTS", "WORKBENCH_REVISIONS", "KNOWLEDGE_USAGE", "POINTACCOUNT"): self.assertNotIn("INSERT INTO " + table, sql) def test_conflicting_old_qc_run_is_preserved_but_not_displayed_as_current_qc(self): self.change("UPDATE qcreport SET run_id=999") plan = self.plan() report = next(i for i in plan["history"] if i["source_kind"] == "qcreport") self.assertEqual(report["source"]["run_id"], 999) self.assertIn("source_run_campaign_mismatch_ledger_only", report["relation_review"]) self.assertEqual(self.resource(plan, "campaign")["data"]["segments"][0]["review"], "") self.assertEqual(self.resource(plan, "campaign")["data"]["review"], "") def test_unrelated_legacy_sessions_are_not_copied(self): self.change("UPDATE agentsession SET campaign_id=999") plan = self.plan() self.assertFalse(any(i["source_kind"] == "agentsession" for i in plan["history"])) self.assertEqual(plan["excluded_history_counts"]["agentsession"], 1) def test_all_historical_artifact_versions_are_preserved(self): older = self.root / "seg_1_r0.md" older.write_text("Previous version", encoding="utf-8") self.change("UPDATE segment SET generation_history=?", (json.dumps([{"round": 0, "artifact_path": str(older)}]),)) plan = self.plan() self.assertEqual(len(plan["artifacts"]), 2) self.assertEqual(plan["artifacts"][str(older)]["body"], "Previous version") def test_missing_artifact_is_explicit_and_does_not_fake_content(self): self.artifact.unlink() plan = self.plan() self.assertEqual(safe_business_summary(plan)["artifacts_missing"], 1) task = self.resource(plan, "campaign")["data"] self.assertEqual(task["segments"][0]["content"], "") self.assertIn("segment_artifact", task["legacy_import"]["missing_configuration"]) def test_artifact_cannot_escape_allowed_directory(self): self.change("UPDATE segment SET artifact_path='/etc/passwd'") with self.assertRaisesRegex(ValueError, "outside"): self.plan() def test_report_omits_private_bodies_and_paths(self): summary = json.dumps(safe_business_summary(self.plan())) for forbidden in ("Original", "Private", "O'Brien", self.tmp.name, "historical prompt", "Node content"): self.assertNotIn(forbidden, summary) def test_tampering_projection_or_artifact_fails(self): mutations = [lambda p: p.update(target_database="production"), lambda p: p.update(target_tenant_id=1), lambda p: p["resources"][0]["target"].update(owner_id=ADMIN_TARGET_ID), lambda p: p["resources"][0]["target"]["data"]["legacy_import"].update(read_only=False), lambda p: p["artifacts"][str(self.artifact)].update(body="Changed body"), lambda p: p["history"][0].update(target_id=AGENT)] for mutate in mutations: plan = self.plan() mutate(plan) with self.subTest(mutate=mutate), self.assertRaises(ValueError): business_sql(plan) def test_bad_fingerprint_fails(self): for digest in (None, "", "X" * 64, "a" * 63): plan = self.plan() plan["source_snapshot_sha256"] = digest with self.subTest(digest=digest), self.assertRaises(ValueError): business_sql(plan) def test_null_in_archived_evidence_is_losslessly_encoded_for_postgres(self): original = "source\x00evidence\ud800" self.change("UPDATE trainingmap SET evidence=?", (json.dumps([{"text": original}]),)) plan = self.plan() encoded = postgres_plan(plan) item = next(r for r in encoded["resources"] if r["source_kind"] == "trainingmap") source = item["source"]["evidence"][0]["text"] self.assertEqual(base64.b64decode(source["body"]).decode("utf-8", errors="surrogatepass"), original) self.assertEqual(item["target"]["data"]["legacy_history"]["evidence"][0]["text"], source) self.assertEqual(self.resource(plan, "trainingmap")["data"]["legacy_history"]["evidence"][0]["text"], original) self.assertNotIn("\\u0000", business_sql(plan)) self.assertNotIn("\\ud800", business_sql(plan)) def test_sql_is_insert_only_tenant_guarded_and_replay_preserves_edits(self): sql = business_sql(self.plan()) for marker in ("current_database()", "yawan_native_acceptance_20260915", "pg_advisory_xact_lock(10059", "Business owner durable identity", "Business Agent durable mapping", "Business durable mapping conflict", "refusing recreation", "Native business ID or owner/name collision", "source_snapshot jsonb", "artifact_snapshot jsonb", "SET LOCAL log_min_error_statement = 'panic'", LEDGER): self.assertIn(marker, sql) for forbidden in ("UPDATE WORKBENCH_RESOURCES", "DELETE FROM", "ON CONFLICT", "COMMIT;", "BEGIN TRANSACTION"): self.assertNotIn(forbidden, sql.upper()) self.assertIn("O''Brien", sql) def test_namespaced_ids_and_plan_are_deterministic(self): self.assertEqual(self.plan(), self.plan()) self.assertNotEqual(business_id("segment", 1), business_id("canvas", 1)) self.assertNotEqual(business_id("trainingmap", 1, "1.0"), business_id("trainingmap", 1, "1.1")) def test_duplicate_node_or_dangling_edge_fails(self): self.change("UPDATE canvas SET edges=?", (json.dumps([{"from": "a", "to": "missing"}]),)) with self.assertRaisesRegex(ValueError, "missing node"): self.plan() if __name__ == "__main__": unittest.main()