xinmi/scripts/test_weknora_business_etl.py

"""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()