xinmi/scripts/test_weknora_identity_etl.py

"""Read-only planner and generated-SQL contract checks for identity migration."""
import copy
import json
import sqlite3
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from weknora_identity_etl import (ADMIN_TARGET_ID, IDENTITY_LEDGER, ORGANIZATION_LEDGER,
                                 PREFERENCE_KEY, TARGET_TENANT, _stable_id, identity_plan,
                                 identity_sql, identity_verification_sql)


class IdentityTests(unittest.TestCase):
    def setUp(self):
        self.db = sqlite3.connect(":memory:")
        self.addCleanup(self.db.close)
        self.db.executescript("""
            CREATE TABLE organization (id INTEGER, name TEXT, slug TEXT, enabled INTEGER);
            CREATE TABLE roledefinition (id INTEGER, role TEXT, label TEXT, description TEXT,
                                         permissions TEXT, builtin INTEGER, organization_id INTEGER, updated_at TEXT);
            CREATE TABLE appuser (id INTEGER, username TEXT, display_name TEXT, role TEXT,
                                  organization_id INTEGER, enabled INTEGER, password_hash TEXT);
            INSERT INTO organization VALUES (1, 'source org', 'default', 1);
            INSERT INTO roledefinition VALUES (1, 'admin', 'Admin', 'All', '["*"]', 1, 1, '2026-09-15');
            INSERT INTO roledefinition VALUES (2, 'editor', 'Editor', 'Edit', '["kb:read", "feature:knowledge", "tasks:write"]', 1, 1, '2026-09-15');
            INSERT INTO roledefinition VALUES (3, 'reviewer', 'Reviewer', 'Review', '["audit:read", "review"]', 1, 1, '2026-09-15');
            INSERT INTO roledefinition VALUES (4, 'viewer', 'Viewer', 'View', '["read"]', 1, 1, '2026-09-15');
            INSERT INTO appuser VALUES (1, 'yawan_admin', 'Legacy Admin', 'admin', 1, 1, 'secret-old-hash');
            INSERT INTO appuser VALUES (3, 'editor_one', 'Display One', 'editor', 1, 1, 'secret-old-hash');
            INSERT INTO appuser VALUES (8, 'wuzong', 'Display Two', 'editor', 1, 1, 'secret-old-hash');
            INSERT INTO appuser VALUES (5, 'deleted_test', 'Test', 'viewer', 1, 0, 'secret-old-hash');
            INSERT INTO appuser VALUES (6, 'deleted_test_two', 'Test Two', 'viewer', 1, 0, 'secret-old-hash');
            INSERT INTO appuser VALUES (7, 'deleted_test_three', 'Test Three', 'viewer', 1, 0, 'secret-old-hash');
        """)

    def plan(self):
        plan = identity_plan(self.db, TARGET_TENANT)
        plan["source_snapshot_sha256"] = "a" * 64
        return plan

    def test_three_enabled_users_and_disabled_manifest(self):
        plan = self.plan()
        self.assertEqual([u["source_user"]["id"] for u in plan["users"]], [1, 3, 8])
        self.assertEqual(plan["skipped"][0]["reason"], "source_account_disabled")
        self.assertEqual(plan["skipped"][0]["source_user"]["id"], 5)
        self.assertEqual(len(plan["skipped"]), 3)
        self.assertNotIn("secret-old-hash", json.dumps(plan))
        admin, first, second = plan["users"]
        self.assertEqual(admin["target_id"], ADMIN_TARGET_ID)
        self.assertEqual(admin["action"], "map_existing_admin")
        for user in (first, second):
            self.assertFalse(user["is_active"])
            self.assertFalse(user["is_system_admin"])
            self.assertFalse(user["can_access_all_tenants"])
            self.assertEqual(user["membership_status"], "suspended")
            self.assertEqual(user["target_membership_role"], "contributor")
            self.assertTrue(user["target_email"].endswith("@migration.invalid"))
            self.assertIn("permission_review_required", user["pending_activation"])
            self.assertIn("password_reset_required", user["pending_activation"])
        metadata = first["preferences"][PREFERENCE_KEY]
        self.assertEqual(metadata["source_user"]["display_name"], "Display One")
        self.assertIn("tasks:write", metadata["source_user"]["permissions"])

    def test_all_source_roles_preserved_including_roles_without_active_users(self):
        roles = {role["role"]: role for role in self.plan()["source_role_definitions"]}
        self.assertEqual(set(roles), {"admin", "editor", "reviewer", "viewer"})
        self.assertEqual(roles["reviewer"]["permissions"], ["audit:read", "review"])
        self.assertEqual(roles["reviewer"]["description"], "Review")
        self.assertEqual(roles["reviewer"]["updated_at"], "2026-09-15")

    def test_sql_requires_snapshot_fingerprint(self):
        for digest in (None, "", "not-a-digest", "A" * 64, "a" * 63):
            plan = self.plan()
            if digest is None:
                del plan["source_snapshot_sha256"]
            else:
                plan["source_snapshot_sha256"] = digest
            with self.subTest(digest=digest), self.assertRaises(ValueError):
                identity_sql(plan)

    def test_planner_does_not_read_password_column_or_write(self):
        def authorize(action, first, second, database, trigger):
            if action == sqlite3.SQLITE_READ and first == "appuser" and second == "password_hash":
                return sqlite3.SQLITE_DENY
            return sqlite3.SQLITE_OK
        self.db.set_authorizer(authorize)
        self.db.execute("PRAGMA query_only=ON")
        self.assertEqual(len(self.plan()["users"]), 3)

    def test_ids_are_deterministic_and_tenant_scoped(self):
        self.assertEqual(self.plan(), self.plan())
        self.assertNotEqual(_stable_id(3, TARGET_TENANT), _stable_id(3, TARGET_TENANT + 1))
        self.assertNotEqual(_stable_id(3, TARGET_TENANT), _stable_id(8, TARGET_TENANT))

    def test_target_tenant_rejected(self):
        for tenant in (1, TARGET_TENANT + 1, str(TARGET_TENANT), True):
            with self.subTest(tenant=tenant), self.assertRaises(ValueError):
                identity_plan(self.db, tenant)

    def test_foreign_source_organization_rejected(self):
        self.db.execute("INSERT INTO organization VALUES (2, 'other', 'other', 1)")
        with self.assertRaisesRegex(ValueError, "organization"):
            self.plan()

    def test_cross_organization_user_rejected_even_if_disabled(self):
        self.db.execute("UPDATE appuser SET organization_id=2 WHERE id=5")
        with self.assertRaisesRegex(ValueError, "organization"):
            self.plan()

    def test_cross_organization_role_rejected(self):
        self.db.execute("UPDATE roledefinition SET organization_id=2 WHERE role='editor'")
        with self.assertRaisesRegex(ValueError, "organization"):
            self.plan()

    def test_unreviewed_enabled_account_rejected(self):
        self.db.execute("INSERT INTO appuser VALUES (10, 'extra', 'Extra', 'editor', 1, 1, '')")
        with self.assertRaisesRegex(ValueError, "reviewed migration batch"):
            self.plan()

    def test_admin_mapping_cannot_silently_change(self):
        self.db.execute("UPDATE appuser SET username='other_admin' WHERE id=1")
        with self.assertRaisesRegex(ValueError, "explicitly mapped"):
            self.plan()

    def test_unknown_enabled_role_rejected(self):
        self.db.execute("UPDATE appuser SET role='viewer' WHERE id=3")
        with self.assertRaisesRegex(ValueError, "Unsupported"):
            self.plan()

    def test_duplicate_source_names_rejected(self):
        self.db.execute("UPDATE appuser SET username='EDITOR_ONE' WHERE id=8")
        with self.assertRaisesRegex(ValueError, "Conflicting"):
            self.plan()

    def test_sql_rejects_tampered_scope_or_activation(self):
        base = self.plan()
        mutations = [
            lambda p: p.update(target_tenant_id=1),
            lambda p: p.update(source="other"),
            lambda p: p.update(target_database="production"),
            lambda p: p["users"][0].update(target_id="other-admin"),
            lambda p: p["users"][1].update(is_active=True),
            lambda p: p["users"][1].update(is_system_admin=True),
            lambda p: p["users"][1].update(can_access_all_tenants=True),
            lambda p: p["users"][1].update(membership_status="active"),
            lambda p: p["users"][1].update(target_membership_role="owner"),
            lambda p: p["users"][1].update(target_email="real@example.com"),
            lambda p: p["users"][1]["source_user"].update(organization_id=2),
            lambda p: p["users"][1]["preferences"][PREFERENCE_KEY].update(source="other"),
        ]
        for mutation in mutations:
            plan = copy.deepcopy(base)
            mutation(plan)
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                identity_sql(plan)

    def test_sql_escapes_source_text_and_does_not_emit_a_password(self):
        self.db.execute("UPDATE appuser SET display_name=? WHERE id=3", ("O'Brien\\note $identity$",))
        sql = identity_sql(self.plan())
        self.assertIn("O''Brien", sql)
        self.assertIn("SET LOCAL standard_conforming_strings = on", sql)
        self.assertIn("gen_random_uuid()", sql)
        self.assertNotIn("secret-old-hash", sql)

    def test_sql_contract_aborts_conflicts_and_preserves_exact_reruns(self):
        sql = identity_sql(self.plan())
        self.assertIn("current_database()", sql)
        self.assertIn("lower(btrim(username))", sql)
        self.assertIn("lower(btrim(email))", sql)
        self.assertIn("ID/provenance collision", sql)
        self.assertIn("username/email collision", sql)
        self.assertIn("Orphan or deleted membership conflicts", sql)
        self.assertIn("existing_user.preferences->'yawan_identity_migration' IS DISTINCT FROM", sql)
        self.assertIn("existing_member.status NOT IN ('suspended', 'active')", sql)
        self.assertIn("unexpected cross-tenant membership", sql)
        self.assertIn("Durable identity mapping conflict", sql)
        self.assertIn("IF NOT has_mapping AND (", sql)
        self.assertIn("refusing to recreate a deleted account", sql)
        self.assertNotIn("ON CONFLICT", sql.upper())
        self.assertNotIn("UPDATE USERS", sql.upper())
        self.assertNotIn("UPDATE TENANT_MEMBERS", sql.upper())
        self.assertNotIn("COMMIT", sql.upper())
        self.assertNotIn("BEGIN TRANSACTION", sql.upper())
        self.assertEqual(sql.count("INSERT INTO users"), 1)
        self.assertEqual(sql.count("INSERT INTO tenant_members"), 1)

    def test_sql_stores_all_identity_mappings_without_modifying_existing_accounts(self):
        sql = identity_sql(self.plan())
        self.assertIn(f"CREATE TABLE IF NOT EXISTS {ORGANIZATION_LEDGER}", sql)
        self.assertIn(f"CREATE TABLE IF NOT EXISTS {IDENTITY_LEDGER}", sql)
        self.assertIn("source_role_definitions jsonb NOT NULL", sql)
        self.assertIn("source_snapshot_sha256 varchar(64) NOT NULL", sql)
        self.assertEqual(sql.count(f"INSERT INTO {IDENTITY_LEDGER}"), 1)
        self.assertGreater(sql.index(f"INSERT INTO {IDENTITY_LEDGER}"), sql.index("Pinned administrator"))
        self.assertNotIn("CONTINUE;", sql)
        self.assertNotIn("ON CONFLICT", sql.upper())

    def test_readonly_verifier_reports_mappings_and_current_state_without_password(self):
        sql = identity_verification_sql(self.plan())
        for field in ("database_matches", "organization_mapping_matches", "identity_mapping_matches",
                      "account_scope_matches", "membership_matches", "is_active", "membership_status"):
            self.assertIn(field, sql)
        for operation in ("INSERT INTO", "UPDATE USERS", "CREATE TABLE", "DELETE FROM", "password_hash"):
            self.assertNotIn(operation, sql)


if __name__ == "__main__":
    unittest.main()