diff --git a/.github/workflows/upgrade-release.yml b/.github/workflows/upgrade-release.yml index ea873141a3..f04b5fdfe0 100644 --- a/.github/workflows/upgrade-release.yml +++ b/.github/workflows/upgrade-release.yml @@ -80,10 +80,11 @@ jobs: VERSION: ${{ github.event.release.tag_name }} run: make release-check - - name: Verify upgrade policy and gate regressions - run: | - make upgrade-policy-check - uv run pytest -q tests/test_upgrade_release_gate.py tests/test_upgrade_publication.py + - name: Run upgrade gate regressions + run: uv run pytest -q tests/test_upgrade_release_gate.py tests/test_upgrade_publication.py tests/test_upgrade_schema_equivalence.py + + - name: Verify upgrade policy + run: make upgrade-policy-check - name: Build release distributions run: | diff --git a/scripts/upgrade_database_probe.py b/scripts/upgrade_database_probe.py index 6e85bb0031..4fa141e17e 100644 --- a/scripts/upgrade_database_probe.py +++ b/scripts/upgrade_database_probe.py @@ -163,7 +163,9 @@ async def inspect(backend: str, database: Path, *, add_revision: bool, canonical trigger_rows = [ dict(zip((column[0] for column in cursor.description), row, strict=True)) for row in cursor.fetchall() ] - triggers = {row["Trigger"]: [row["Event"], row["Table"], row["Statement"]] for row in trigger_rows} + triggers = { + row["Trigger"]: [row["Timing"], row["Event"], row["Table"], row["Statement"]] for row in trigger_rows + } logical = None if canonical: from powercontext.builtin.persistence.migrations.mysql_schema import _parse_table diff --git a/scripts/upgrade_release_gate.py b/scripts/upgrade_release_gate.py index f6ef42ff5e..da4dfe2ce6 100644 --- a/scripts/upgrade_release_gate.py +++ b/scripts/upgrade_release_gate.py @@ -390,6 +390,7 @@ def case_workspace(backend: str, output: Path) -> Iterator[Path]: def run_case(root: Path, wheel: Path, case: dict[str, Any], output: Path) -> dict[str, Any]: + require(sys.platform == "linux", "Installed-release acceptance requires Linux process ownership and cleanup") output.mkdir(parents=True, exist_ok=False) # Never inherit a developer's live deployment or pytest selection overrides. environment = { diff --git a/tests/release_upgrades/README.md b/tests/release_upgrades/README.md index 6197291be2..d06e17e452 100644 --- a/tests/release_upgrades/README.md +++ b/tests/release_upgrades/README.md @@ -82,7 +82,7 @@ those two upgrade configurations alone, or `full` to run all 12. Other Main CI s For a PR branch, the equivalent command is: ```sh -gh workflow run master.yml --repo oceanbase/powercontext --ref codex/issue-2154-full-server-upgrade -f upgrade_selection=seekdb-code +gh workflow run master.yml --repo oceanbase/powercontext --ref YOUR_BRANCH -f upgrade_selection=seekdb-code ``` `automatic` runs the remaining 10. Reports list deferred cases and mark full acceptance incomplete; a passing @@ -125,12 +125,17 @@ Provide these behavioral tests (additional tests are allowed): | `test_public_migration_commands` | Stop writers, run the target artifact's public maintenance commands, and validate full readiness. SQLite additionally restores the actual automatic recovery point, reads it with the source Server, upgrades it and verifies target use. A limited bundle verification does not pass. | | `test_target_startup` | Start the target Server against the migrated database and verify readiness. | | `test_read_write_and_retrieval` | Read the old records, write new ones, and retrieve both through public interfaces. | -| `test_authorization` | Verify permitted and denied access and preservation of relevant grants. | +| `test_authorization` | Preserve the old viewer grant; verify allowed reads, denied cross-scope reads, denied writes and unchanged data after rejection using both installed releases. Invalid credentials must still fail. | | `test_candidates_and_processing` | Verify retained Candidates and CAS approval; load and acknowledge the old pending intent using the target installed SDK, including idempotency. | | `test_background_processing` | Run the installed target's actual Memory Worker against a local controlled inference endpoint; capture a new Source, verify the published Memory's exact Source reference and FTS retrieval, and require the child Worker's usage report. | | `test_preserved_history_and_cursors` | Verify exact history/references and unchanged or declared-transformed cursor and receipt semantics. | | `test_fresh_upgrade_equivalence` | Initialize a fresh target database; compare schema/backend objects and behavior with the upgraded database. The current suite has no retained/archive object exclusions. | +The authorization probe runs each installed release's native Server through real HTTP handlers with a fixed +ordinary-user authentication provider. The persisted builtin authorization policy is not overridden. Both the +source and target CLI Servers are stopped before the SDK probe opens storage, so embedded seekDB has one owner. +CLI readiness remains a separate mandatory phase. + Use a session fixture for the shared lifecycle; do not depend on pytest's test execution order. All collected tests must execute and pass. Empty collection, missing phases, skips (including missing backends), xfail, failure, command failure, and missing/malformed reports fail the gate. A real OceanBase suite must provision an isolated database and @@ -153,7 +158,8 @@ The separate SDK fixture tests the old pending intent's format and acknowledgmen execution and evidence preservation, not real-model quality. Vector storage uses its separately registered profiles; passing a no-vector case cannot establish vector acceptance. SQLite inventories include indexes, views and triggers; MySQL-family table DDL includes indexes and constraints, -and views are collected separately. MySQL comparison parses columns, indexes (including parser properties), constraints, and CHECK expressions, ignoring only physical formatting and backend-generated constraint names. Views and triggers are collected independently. SQLite normalizes whitespace and identifier quoting. Structural differences fail rather than being silently excluded. +and views are collected separately. MySQL comparison parses columns, indexes (including parser properties), constraints, and CHECK expressions, ignoring only physical formatting and backend-generated constraint names. Views and triggers are collected independently; native trigger comparison includes timing (`BEFORE`/`AFTER`), +event, table and statement. SQLite preserves quoted tokens and normalizes only unquoted formatting whitespace. Structural differences fail rather than being silently excluded. Regenerate installation support regions with `uv run python scripts/upgrade_release_gate.py --write-docs` and run `make upgrade-policy-check`. When persistence @@ -206,3 +212,31 @@ Main first runs the previously failing embedded seekDB concurrent-startup regres once, before the installed-release workflow and the remaining seekDB clean-migration job. Those jobs still run after a failed priority check unless the workflow is cancelled; the selected installed-release cases remain disjoint and parallel. Other CI suites can run concurrently. + +## Regression checks + +The shared prepare job runs gate, publication and schema-equivalence regressions before checking the reviewed +persistence digest. Digest failure still blocks preparation; the earlier regression result remains visible. +The raw DDL comparator preserves quoted literals, comments and identifier boundaries, normalizing only unquoted +whitespace and table-level auto-increment counters. Native logical table reflection remains authoritative for +seekDB/OceanBase comparisons, with views and triggers also compared. SQL literal differences must never disappear +as formatting changes. Unknown configurations and missing or mismatched execution evidence fail closed. + +Full installed-release acceptance requires Linux process ownership and descendant cleanup. Other hosts can run +the policy and comparison regressions, but the full runner rejects them before creating workspaces or installing +packages. Passing local regressions is not evidence of a successful native-backend upgrade. + + +## Acceptance boundaries + +A passing run establishes the selected cases for that exact target build, not every historical production database. +The source fixture is newly initialized with the official 1.3.0 package and representative data; it does not recreate +an older-version-to-1.3.0 lineage or an existing user's database. Each future target build must rerun acceptance. +The Linux matrix does not establish macOS embedded-engine, desktop-daemon or in-place tool-installation behavior. +Source and target dependency resolution does not freeze the dependencies of a historical installation, so recorded +engine/driver versions must be considered when interpreting compatibility evidence. + +The positive path exercises the public `db-migrate` commands, not the historical `atomic-memory-migrate` procedure. +SQLite tests automatic backup restoration; seekDB and OceanBase explicitly skip backups in this suite and do not +establish backup/rollback acceptance. The two manual embedded seekDB Code cases require their own successful runs; +the ten automatic results cannot establish those configurations. diff --git a/tests/release_upgrades/authorization_fixture.py b/tests/release_upgrades/authorization_fixture.py new file mode 100644 index 0000000000..6330623a35 --- /dev/null +++ b/tests/release_upgrades/authorization_fixture.py @@ -0,0 +1,88 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Check real HTTP authorization with each installed release's native SDK. + +Run with -I after stopping the CLI Server. Only authentication is supplied; +the installed Server loads and enforces the persisted builtin access policy. +""" + +import asyncio +import json +import sys +from pathlib import Path + +import httpx + +from powercontext.server.authentication import StaticBearerAuthenticationProvider +from powercontext.server.authz import PrincipalRef +from powercontext.server.factory import create_server_app +from powercontext.server.settings import ServerSettings + + +async def verify(historical): + token = "isolated-viewer-token" # noqa: S105 -- disposable fixture credential + viewer = PrincipalRef(type="user", id="upgrade-viewer") + settings = ServerSettings() + # Keep the persisted processing capabilities unchanged. Embedded releases + # require the all role; use a separate, unprivileged background identity + # and disable scheduling instead of running workers as the viewer. + settings = settings.model_copy( + update={ + "access": settings.access.model_copy(update={"background_principal_id": "upgrade-authz-probe"}), + "runtime": settings.runtime.model_copy( + update={ + "schedule_seconds": None, + "experience_schedule_seconds": None, + } + ), + } + ) + app = create_server_app( + settings=settings, + authentication_provider=StaticBearerAuthenticationProvider(token, viewer), + ) + async with ( + app.router.lifespan_context(app), + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://fixture", + headers={"Authorization": f"Bearer {token}"}, + ) as client, + ): + identity = await client.get("/v1/access/me") + assert identity.status_code == 200, identity.text + assert identity.json()["principal"]["type"] == "user" + assert identity.json()["principal"]["id"] == viewer.id + scope = historical["scope"] + allowed = await client.get(f"/v1/scopes/{scope}") + assert allowed.status_code == 200, allowed.text + records = await client.post("/v1/memory/entries/list", json={"scope_id": scope}) + assert records.status_code == 200, records.text + assert {item["text"] for item in records.json()["entries"]} >= {item["text"] for item in historical["memory"]} + denied_read = await client.get(f"/v1/scopes/{historical['ungranted_scope']}") + assert denied_read.status_code == 403, denied_read.text + denied_write = await client.post( + "/v1/memory/remember", + json={"scope_id": scope, "kind": "fact", "text": "Viewer must not persist this write."}, + ) + assert denied_write.status_code == 403, denied_write.text + after = await client.post("/v1/memory/entries/list", json={"scope_id": scope}) + assert after.status_code == 200, after.text + assert after.json() == records.json(), "Denied writes must not change stored records" + print("Viewer can read its granted scope; cross-scope reads and writes are denied.") + + +if __name__ == "__main__": + asyncio.run(verify(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")))) diff --git a/tests/release_upgrades/test_full_server.py b/tests/release_upgrades/test_full_server.py index da39ecdfe5..c5462e9b79 100644 --- a/tests/release_upgrades/test_full_server.py +++ b/tests/release_upgrades/test_full_server.py @@ -370,6 +370,17 @@ def task_state(case, role, operation, scope): return json.loads(path.read_text()) +def verify_authorization(case, role, historical): + fixture = case["report"] / "authorization-fixture.json" + fixture.write_text(json.dumps(historical), encoding="utf-8") + command( + [str(case[role]), "-I", str(ROOT / "tests/release_upgrades/authorization_fixture.py"), str(fixture)], + environment_for(case), + case["work"], + case["report"] / f"{role}-authorization.log", + ) + + def seed(client, *, candidate_prefix): scopes = [ post( @@ -423,9 +434,8 @@ def seed(client, *, candidate_prefix): {"scope_id": scope, "candidate_id": published["candidate_id"], "expected_version": published["version"]}, ) artifact = post(client, "/v1/experience/get", {"scope_id": scope, "artifact": published["result_artifact"]}) - principal = client.get("/v1/access/me").json()["principal"] grant = { - "subject": principal, + "subject": {"type": "user", "id": "upgrade-viewer"}, "resource": {"type": "scope", "scope_id": scope}, "role": "scope.viewer", "idempotency_key": "upgrade-viewer", @@ -440,6 +450,7 @@ def seed(client, *, candidate_prefix): assert {digest(item) for item in readback} >= {digest(item) for item in memory} return { "scope": scope, + "ungranted_scope": scopes[1]["scope_id"], "memory": memory, "candidate": candidate, "first_candidate": first, @@ -547,6 +558,7 @@ def historical(case): }, ) assert_vector_reads(client, value) + verify_authorization(case, "source", value) if case["code_enabled"]: value["code"] = historical_code(case, value["scope"]) value["task"] = task_state(case, "source", "seed", value["scope"]) @@ -960,12 +972,17 @@ def test_read_write_and_retrieval(case, target, historical): assert_vector_reads(target, historical, new_text=text) -def test_authorization(target, historical): - assert target.get("/v1/scopes", headers={"Authorization": "Bearer invalid-fixture-token"}).status_code == 401 - # The stable idempotency key must resolve to the SAME persisted grant. - assert post(target, "/v1/access/bindings/create", historical["grant"]) == historical["binding"] - response = target.get(f"/v1/scopes/{historical['scope']}") - assert response.status_code == 200 +def test_authorization(case, migrated): + historical = migrated["historical"] + with server( + case["target"], environment_for(case), case["work"], case["report"] / "authorization-server.log" + ) as client: + assert client.get("/v1/scopes", headers={"Authorization": "Bearer invalid-fixture-token"}).status_code == 401 + # The stable idempotency key must resolve to the SAME persisted grant. + assert post(client, "/v1/access/bindings/create", historical["grant"]) == historical["binding"] + # Stop the CLI before opening the same database with the installed SDK, + # especially for the embedded engine's exclusive process ownership. + verify_authorization(case, "target", historical) def test_candidates_and_processing(case, migrated): @@ -1025,6 +1042,17 @@ def test_preserved_history_and_cursors(target, historical): assert stored[key] == version[key] +# Quoted tokens and comments are opaque: whitespace, quote characters and +# AUTO_INCREMENT text inside them are schema content, not formatting. +DDL_TOKENS = re.compile( + r"""'(?:''|\\.|[^'\\])*' | "(?:""|\\.|[^"\\])*" | `(?:``|\\.|[^`\\])*` + | \[(?:\]\]|[^\]])*\] | /\*.*?\*/ | --[^\r\n]* + | [A-Za-z_\u0080-\U0010ffff][A-Za-z_0-9$\u0080-\U0010ffff]* | [0-9]+ | ->> | -> | <= | >= | <> | != | \|\| | << | >> | [^\s] + """, + re.VERBOSE | re.DOTALL, +) + + def canonical_schema(snapshot): if snapshot.get("logical_tables") is not None: return { @@ -1034,9 +1062,35 @@ def canonical_schema(snapshot): } def normalize(sql): - sql = re.sub(r"\bAUTO_INCREMENT=\d+", "", sql) - sql = re.sub(r'["`]([A-Za-z_][A-Za-z_0-9]*)["`]', r"\1", sql) - return " ".join(sql.split()) + tokens = DDL_TOKENS.findall(sql) + table = [lexeme.upper() for lexeme in tokens[:2]] == ["CREATE", "TABLE"] + depth = 0 + closed_definition = False + normalized = [] + index = 0 + while index < len(tokens): + lexeme = tokens[index] + # Only ignore the table-level next-id counter after the definition. + # Column AUTO_INCREMENT attributes and all quoted contents remain. + if ( + table + and closed_definition + and depth == 0 + and lexeme.upper() == "AUTO_INCREMENT" + and tokens[index + 1 : index + 2] == ["="] + and index + 2 < len(tokens) + and tokens[index + 2].isdigit() + ): + index += 3 + continue + if lexeme == "(": + depth += 1 + elif lexeme == ")": + depth -= 1 + closed_definition = depth == 0 + normalized.append(lexeme) + index += 1 + return tuple(normalized) return { kind: {name: normalize(sql or "") for name, sql in snapshot.get(kind, {}).items()} diff --git a/tests/test_upgrade_release_gate.py b/tests/test_upgrade_release_gate.py index 8377f2cf9b..ae6bb5fabb 100644 --- a/tests/test_upgrade_release_gate.py +++ b/tests/test_upgrade_release_gate.py @@ -1205,3 +1205,69 @@ def test_serial_publication_gate_also_waives_only_seekdb_code(release): assert result["complete"] is False assert all(gate.manual_upgrade_case(case) for case in result["deferred_cases"]) assert all(not gate.manual_upgrade_case(item["case"]) for item in result["cases"]) + + +@pytest.mark.parametrize("configuration", ["typo"]) +def test_unknown_configuration_fails_before_opening_database(tmp_path, configuration): + work = tmp_path / "must-not-be-created" + suite = gate.ROOT / "tests/release_upgrades/test_full_server.py" + environment = dict(os.environ) + environment.update( + PYTEST_DISABLE_PLUGIN_AUTOLOAD="1", + PC_UPGRADE_CONFIGURATION=configuration, + PC_UPGRADE_SOURCE_VERSION="1.3.0", + PC_UPGRADE_SOURCE_PYTHON=str(tmp_path / "nonexistent-source"), + PC_UPGRADE_TARGET_PYTHON=str(tmp_path / "nonexistent-target"), + PC_UPGRADE_BACKEND="sqlite", + PC_UPGRADE_WORK=str(work), + PC_UPGRADE_REPORT_DIR=str(tmp_path / "report"), + ) + environment.pop("PYTEST_ADDOPTS", None) + result = subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-c", + os.devnull, + "--confcutdir", + str(suite.parent), + "-p", + "no:cacheprovider", + "-k", + "test_source_release_database", + "-q", + str(suite), + ], + cwd=tmp_path, + env=environment, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + assert result.returncode != 0 + assert "Unregistered acceptance configuration" in result.stdout + assert not work.exists() + + +@pytest.mark.skipif(sys.platform == "linux", reason="Linux supports the required process ownership") +def test_full_acceptance_rejects_unsupported_host_before_installing(tmp_path): + with pytest.raises(gate.GateError, match="requires Linux"): + gate.run_case(tmp_path, tmp_path / "wheel", {}, tmp_path / "report") + assert not (tmp_path / "report").exists() + + +@pytest.mark.parametrize("configuration", [None, "different-profile"]) +def test_parallel_collection_rejects_unproven_configuration(parallel_reports, configuration): + root, prepared, reports, _ = parallel_reports + path = reports / "case-0/acceptance/backend.json" + backend = json.loads(path.read_text()) + if configuration is None: + del backend["configuration"] + else: + backend["configuration"] = configuration + parallel.write_json(path, backend) + with pytest.raises(gate.GateError, match="Configuration evidence"): + parallel.collect(root, prepared, reports, root / "accepted") + assert not (root / "accepted/validated-dist").exists() diff --git a/tests/test_upgrade_schema_equivalence.py b/tests/test_upgrade_schema_equivalence.py new file mode 100644 index 0000000000..863af30316 --- /dev/null +++ b/tests/test_upgrade_schema_equivalence.py @@ -0,0 +1,136 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Prevent schema equivalence from erasing differences in SQL literals.""" + +import asyncio +import json +import sqlite3 +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from scripts import upgrade_database_probe +from tests.release_upgrades.test_full_server import canonical_schema + + +def schema(sql): + return canonical_schema({"tables": {"example": sql}}) + + +@pytest.mark.parametrize( + "left,right", [("a b", "a b"), ('"word"', "word"), ("AUTO_INCREMENT=12", "AUTO_INCREMENT=99")] +) +def test_distinct_sqlite_defaults_do_not_compare_equal(left, right): + inventories = [] + values = [] + for default in (left, right): + with sqlite3.connect(":memory:") as connection: + connection.execute(f"CREATE TABLE example (value TEXT DEFAULT '{default}')") + ddl = connection.execute("SELECT sql FROM sqlite_schema WHERE name='example'").fetchone()[0] + connection.execute("INSERT INTO example DEFAULT VALUES") + values.append(connection.execute("SELECT value FROM example").fetchone()[0]) + inventories.append(schema(ddl)) + assert values[0] != values[1] + assert inventories[0] != inventories[1] + + +def test_distinct_sqlite_checks_do_not_compare_equal(): + inventories = [] + for literal in ("a b", "a b"): + with sqlite3.connect(":memory:") as connection: + connection.execute(f"CREATE TABLE example (value TEXT CHECK (value = '{literal}'))") + connection.execute("INSERT INTO example VALUES (?)", (literal,)) + with pytest.raises(sqlite3.IntegrityError): + connection.execute("INSERT INTO example VALUES (?)", ("a b" if literal == "a b" else "a b",)) + inventories.append( + schema(connection.execute("SELECT sql FROM sqlite_schema WHERE name='example'").fetchone()[0]) + ) + assert inventories[0] != inventories[1] + + +def test_table_counters_and_unquoted_whitespace_are_not_schema_changes(): + left = "CREATE TABLE `example` (`id` int AUTO_INCREMENT, `value` text DEFAULT 'a b') AUTO_INCREMENT=12" + right = "CREATE TABLE `example` ( `id` int AUTO_INCREMENT, `value` text DEFAULT 'a b' ) AUTO_INCREMENT = 99" + assert schema(left) == schema(right) + assert schema(left) != schema(left.replace("int AUTO_INCREMENT", "int")) + + +@pytest.mark.parametrize("literal", [r"'a\' b'", "'it''s quoted'", '"a b"', "`a b`", "[a b]"]) +def test_quoted_tokens_preserve_spaces_and_escapes(literal): + ddl = f"CREATE TABLE example (value TEXT DEFAULT {literal})" + assert schema(ddl) != schema(ddl.replace(" ", " ")) + + +@pytest.mark.parametrize("name", ["字段", "😀😀"]) +def test_unquoted_identifier_boundaries_are_preserved(name): + inventories = [] + columns = [] + for declaration in (name, " ".join(name)): + with sqlite3.connect(":memory:") as connection: + connection.execute(f"CREATE TABLE example ({declaration} TEXT)") + columns.append(connection.execute("PRAGMA table_info(example)").fetchone()[1]) + inventories.append( + schema(connection.execute("SELECT sql FROM sqlite_schema WHERE name='example'").fetchone()[0]) + ) + assert columns[0] != columns[1] + assert inventories[0] != inventories[1] + + +@pytest.mark.parametrize("kind", ["logical_tables", "views", "triggers"]) +def test_native_logical_schema_preserves_object_differences(kind): + left = { + "logical_tables": {"example": {"default": "a b"}}, + "views": {"example_view": "SELECT 'a b'"}, + "triggers": {"example_trigger": ["BEFORE", "INSERT", "example", "SET NEW.value = 'a b'"]}, + } + right = json.loads(json.dumps(left).replace("a b", "a b")) + changed = {**left, kind: right[kind]} + assert canonical_schema(left) != canonical_schema(changed) + + +def test_native_probe_preserves_trigger_timing(monkeypatch, tmp_path): + """Synthetic driver rows exercise the probe and comparator, not a live backend.""" + monkeypatch.setenv("PC_UPGRADE_OCEANBASE_URL", "mysql://fixture@localhost/fixture") + monkeypatch.delenv("POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL", raising=False) + monkeypatch.setattr(upgrade_database_probe, "version", lambda _name: "test-driver") + snapshots = [] + for timing in ("BEFORE", "AFTER"): + cursor = MagicMock() + cursor.fetchone.return_value = ("test-engine", "mysql") + cursor.fetchall.return_value = [] + + def execute(query, *, timing=timing, cursor=cursor): + if query == "SHOW TRIGGERS": + fields = ("Trigger", "Event", "Table", "Statement", "Timing") + cursor.description = [(field,) for field in fields] + cursor.fetchall.return_value = [("example_trigger", "INSERT", "example", "SET @seen = 1", timing)] + + cursor.execute.side_effect = execute + connection = MagicMock() + connection.cursor.return_value.__enter__.return_value = cursor + connect = MagicMock() + connect.return_value.__enter__.return_value = connection + monkeypatch.setattr( + upgrade_database_probe, "import_module", lambda _name, connect=connect: SimpleNamespace(connect=connect) + ) + snapshot = asyncio.run( + upgrade_database_probe.inspect("oceanbase", tmp_path, add_revision=False, canonical=True) + ) + snapshots.append(json.loads(json.dumps(snapshot))) + + assert snapshots[0]["trigger_rows"][0]["Timing"] == "BEFORE" + assert snapshots[1]["trigger_rows"][0]["Timing"] == "AFTER" + assert canonical_schema(snapshots[0]) != canonical_schema(snapshots[1])