307 lines
11 KiB
Python
307 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ops import run_local_canonical_postgres_rebuild as rebuild
|
|
|
|
|
|
def _docker_available() -> bool:
|
|
if shutil.which("docker") is None:
|
|
return False
|
|
completed = subprocess.run(
|
|
["docker", "info", "--format", "{{.ServerVersion}}"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
return completed.returncode == 0
|
|
|
|
|
|
def _docker_psql(container: str, database: str, sql: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
[
|
|
"docker",
|
|
"exec",
|
|
"-i",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
],
|
|
input=sql,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _wait_for_postgres(container: str, database: str) -> None:
|
|
for _ in range(120):
|
|
completed = _docker_psql(container, database, "select 1;")
|
|
if completed.returncode == 0:
|
|
return
|
|
time.sleep(0.25)
|
|
pytest.fail(f"fixture PostgreSQL did not become ready: {container}")
|
|
|
|
|
|
def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> None:
|
|
if completed.returncode != 0:
|
|
pytest.fail(
|
|
f"{action} failed ({completed.returncode})\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
|
|
)
|
|
|
|
|
|
def _capture_manifest(container: str, database: str, destination: Path) -> None:
|
|
copy = subprocess.run(
|
|
["docker", "cp", str(rebuild.DEFAULT_MANIFEST_SQL), f"{container}:/tmp/parity.sql"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
_require_success(copy, "manifest SQL copy")
|
|
capture = subprocess.run(
|
|
[
|
|
"docker",
|
|
"exec",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
"-f",
|
|
"/tmp/parity.sql",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
_require_success(capture, "source manifest capture")
|
|
destination.write_text(capture.stdout, encoding="utf-8")
|
|
|
|
|
|
def _capture_dump(container: str, database: str, destination: Path) -> None:
|
|
dump = subprocess.run(
|
|
[
|
|
"docker",
|
|
"exec",
|
|
container,
|
|
"pg_dump",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
database,
|
|
"--format=custom",
|
|
"--file=/tmp/canonical.dump",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
_require_success(dump, "fixture pg_dump")
|
|
copy = subprocess.run(
|
|
["docker", "cp", f"{container}:/tmp/canonical.dump", str(destination)],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
_require_success(copy, "fixture dump copy")
|
|
|
|
|
|
def _fixture_sql(database: str) -> str:
|
|
return f"""
|
|
create schema kb_stage;
|
|
create table public.claims (id bigint primary key, body text not null);
|
|
create table public.sources (id bigint primary key, url text not null);
|
|
create table public.claim_evidence (
|
|
id bigint primary key,
|
|
claim_id bigint not null references public.claims(id),
|
|
source_id bigint not null references public.sources(id)
|
|
);
|
|
create table public.claim_edges (
|
|
id bigint primary key,
|
|
from_claim bigint not null references public.claims(id),
|
|
to_claim bigint not null references public.claims(id)
|
|
);
|
|
create table public.reasoning_tools (id bigint primary key, name text not null);
|
|
create table kb_stage.kb_proposals (
|
|
id bigint primary key,
|
|
status text not null,
|
|
created_at timestamptz not null
|
|
);
|
|
insert into public.claims values (1, 'fixture claim');
|
|
insert into public.sources values (1, 'https://example.invalid/source');
|
|
insert into public.claim_evidence values (1, 1, 1);
|
|
insert into public.claim_edges values (1, 1, 1);
|
|
insert into public.reasoning_tools values (1, 'fixture tool');
|
|
insert into kb_stage.kb_proposals values (1, 'pending_review', '2026-07-16T00:00:00Z');
|
|
|
|
create role kb_review nologin noinherit;
|
|
create role kb_apply login inherit;
|
|
create role kb_observatory_read nologin noinherit;
|
|
alter role kb_apply set application_name to 'teleo=parity-fixture';
|
|
alter role kb_apply in database "{database}" set statement_timeout to '17s';
|
|
alter role kb_observatory_read in database "{database}" set default_transaction_read_only to 'on';
|
|
|
|
grant kb_review to kb_apply
|
|
with admin true, inherit false, set false
|
|
granted by postgres;
|
|
set role kb_apply;
|
|
grant kb_review to kb_observatory_read
|
|
with admin false, inherit true, set false
|
|
granted by kb_apply;
|
|
reset role;
|
|
|
|
revoke all privileges on database "{database}" from public cascade;
|
|
grant connect on database "{database}" to public granted by postgres;
|
|
grant connect on database "{database}" to kb_review with grant option granted by postgres;
|
|
set role kb_review;
|
|
grant connect on database "{database}" to kb_apply granted by kb_review;
|
|
reset role;
|
|
"""
|
|
|
|
|
|
@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for PostgreSQL parity")
|
|
def test_real_postgres_rebuild_replays_scoped_settings_memberships_public_acl_and_grantors(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source_container = f"teleo-pr173-source-{uuid.uuid4().hex[:12]}"
|
|
source_database = "teleo_source"
|
|
started = subprocess.run(
|
|
[
|
|
"docker",
|
|
"run",
|
|
"--rm",
|
|
"--detach",
|
|
"--network",
|
|
"none",
|
|
"--name",
|
|
source_container,
|
|
"--tmpfs",
|
|
f"{rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size=256m",
|
|
"--env",
|
|
f"POSTGRES_DB={source_database}",
|
|
"--env",
|
|
"POSTGRES_HOST_AUTH_METHOD=trust",
|
|
rebuild.DEFAULT_IMAGE,
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
_require_success(started, "source PostgreSQL start")
|
|
|
|
cleanup_verified = False
|
|
try:
|
|
_wait_for_postgres(source_container, source_database)
|
|
bootstrap = _docker_psql(source_container, source_database, _fixture_sql(source_database))
|
|
_require_success(bootstrap, "source parity fixture bootstrap")
|
|
|
|
source_manifest_path = tmp_path / "source-manifest.jsonl"
|
|
source_dump_path = tmp_path / "source.dump"
|
|
_capture_manifest(source_container, source_database, source_manifest_path)
|
|
_capture_dump(source_container, source_database, source_dump_path)
|
|
|
|
source_manifest = rebuild.load_manifest(source_manifest_path)
|
|
roles = {row["name"]: row for row in source_manifest["singleton"]["application_roles"]["items"]}
|
|
assert roles["kb_apply"]["settings"] == [{"name": "application_name", "value": "teleo=parity-fixture"}]
|
|
assert roles["kb_apply"]["database_settings"] == [{"name": "statement_timeout", "value": "17s"}]
|
|
assert roles["kb_observatory_read"]["database_settings"] == [
|
|
{"name": "default_transaction_read_only", "value": "on"}
|
|
]
|
|
memberships = source_manifest["singleton"]["role_memberships"]["items"]
|
|
assert {row["grantor"] for row in memberships} == {"postgres", "kb_apply"}
|
|
assert any(
|
|
row["admin_option"] is True and row["inherit_option"] is False and row["set_option"] is False
|
|
for row in memberships
|
|
)
|
|
database_acls = [
|
|
row for row in source_manifest["singleton"]["object_acl"]["items"] if row["object_type"] == "database"
|
|
]
|
|
assert any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "CONNECT" for row in database_acls)
|
|
assert not any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "TEMPORARY" for row in database_acls)
|
|
assert any(row["grantor"] == "kb_review" and row["grantee"] == "kb_apply" for row in database_acls)
|
|
|
|
receipt_path = tmp_path / "parity-receipt.json"
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"ops/run_local_canonical_postgres_rebuild.py",
|
|
"--dump",
|
|
str(source_dump_path),
|
|
"--database",
|
|
"teleo_target",
|
|
"--source-manifest",
|
|
str(source_manifest_path),
|
|
"--manifest-sql",
|
|
str(rebuild.DEFAULT_MANIFEST_SQL),
|
|
"--tmpfs-mb",
|
|
"256",
|
|
"--max-target-query-ms",
|
|
"10000",
|
|
"--max-target-source-ratio",
|
|
"1000",
|
|
"--output",
|
|
str(receipt_path),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
|
|
assert completed.returncode == 0, completed.stdout + completed.stderr
|
|
assert receipt["status"] == "pass", receipt.get("error")
|
|
assert receipt["application_roles_bootstrapped"] == [
|
|
"kb_apply",
|
|
"kb_observatory_read",
|
|
"kb_review",
|
|
]
|
|
assert receipt["role_setting_statement_count"] == 1
|
|
assert receipt["database_role_setting_statement_count"] == 2
|
|
assert receipt["role_membership_statement_count"] == 2
|
|
assert receipt["database_acl_statement_count"] == len(database_acls)
|
|
assert receipt["parity"]["status"] == "pass"
|
|
assert receipt["parity"]["problems"] == []
|
|
structural = receipt["parity"]["details"]["structural_hashes"]
|
|
assert structural["role_memberships"]["source"] == structural["role_memberships"]["target"]
|
|
assert structural["object_acl"]["source"] == structural["object_acl"]["target"]
|
|
application_roles = receipt["parity"]["details"]["application_role_hashes"]
|
|
assert application_roles["source"] == application_roles["target"]
|
|
assert receipt["parity"]["details"]["application_role_mismatches"] == {}
|
|
assert receipt["parity"]["source_manifest"]["sha256"] != receipt["parity"]["target_manifest"]["sha256"]
|
|
assert receipt["cleanup"]["status"] == "pass"
|
|
assert receipt["cleanup"]["container_absent"] is True
|
|
finally:
|
|
subprocess.run(
|
|
["docker", "rm", "--force", source_container],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
inspected = subprocess.run(
|
|
["docker", "container", "inspect", source_container],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
cleanup_verified = inspected.returncode != 0
|
|
|
|
assert cleanup_verified is True
|