sysadmin
2429d9d2e8
Legacy NULL-scope dedupe only compared Gitea targets, so duplicate rows with the same provider key and issue but different fingerprint/status/ event_count/etc. collapsed to the lowest link_id and silently dropped observation data. Migration now compares all meaningful observation fields and refuses to discard conflicts (#619 RC3 / #613).
825 lines
29 KiB
Python
825 lines
29 KiB
Python
"""Tests for control-plane DB substrate (#613)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import timedelta
|
|
|
|
from control_plane_db import (
|
|
ControlPlaneDB,
|
|
InvalidWorkKindError,
|
|
LeaseRequiredError,
|
|
WORK_KINDS,
|
|
_ts,
|
|
_utc_now,
|
|
)
|
|
|
|
|
|
class ControlPlaneDBTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
|
|
self.db = ControlPlaneDB(self.db_path)
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
def test_schema_and_architecture_meta(self) -> None:
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
|
|
finally:
|
|
conn.close()
|
|
self.assertEqual(rows["schema_version"], "2")
|
|
self.assertIn("DB coordinates", rows["architecture"])
|
|
self.assertIn("bridge", rows["architecture"].lower())
|
|
|
|
def test_rejects_raw_incident_as_work_kind(self) -> None:
|
|
with self.assertRaises(InvalidWorkKindError):
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="org",
|
|
repo="repo",
|
|
kind="sentry_incident",
|
|
number=1,
|
|
)
|
|
with self.assertRaises(InvalidWorkKindError):
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="org",
|
|
repo="repo",
|
|
kind="glitchtip_incident",
|
|
number=9,
|
|
)
|
|
self.assertEqual(WORK_KINDS, frozenset({"issue", "pr"}))
|
|
|
|
def test_atomic_assign_and_lease_fields(self) -> None:
|
|
self.db.upsert_session(session_id="s-a", role="author", profile="prgs-author")
|
|
result = self.db.assign_and_lease(
|
|
session_id="s-a",
|
|
role="author",
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
kind="issue",
|
|
number=613,
|
|
expected_head_sha="abc123",
|
|
allowed_actions=("implement", "comment"),
|
|
forbidden_actions=("approve", "merge"),
|
|
)
|
|
self.assertEqual(result.outcome, "assigned")
|
|
self.assertIsNotNone(result.assignment_id)
|
|
self.assertIsNotNone(result.lease_id)
|
|
self.assertEqual(result.role, "author")
|
|
self.assertEqual(result.work_kind, "issue")
|
|
self.assertEqual(result.work_number, 613)
|
|
self.assertEqual(result.expected_head_sha, "abc123")
|
|
self.assertIn("implement", result.allowed_actions)
|
|
self.assertIn("merge", result.forbidden_actions)
|
|
self.assertIsNotNone(result.expires_at)
|
|
|
|
def test_second_session_waits_on_foreign_lease(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
self.db.upsert_session(session_id="s2", role="author")
|
|
first = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=100,
|
|
expected_head_sha="deadbeef",
|
|
)
|
|
self.assertEqual(first.outcome, "assigned")
|
|
second = self.db.assign_and_lease(
|
|
session_id="s2",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=100,
|
|
expected_head_sha="deadbeef",
|
|
)
|
|
self.assertEqual(second.outcome, "wait")
|
|
self.assertEqual(second.owner_session_id, "s1")
|
|
|
|
def test_owner_resume_refreshes_lease(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="reviewer")
|
|
a = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="reviewer",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=50,
|
|
expected_head_sha="head-50",
|
|
)
|
|
b = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="reviewer",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=50,
|
|
expected_head_sha="head-50",
|
|
)
|
|
self.assertEqual(b.outcome, "assigned")
|
|
self.assertEqual(b.lease_id, a.lease_id)
|
|
self.assertIn("owner-resume", b.reason)
|
|
|
|
def test_require_valid_assignment_gates_mutations(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=7,
|
|
allowed_actions=("implement",),
|
|
forbidden_actions=("merge",),
|
|
)
|
|
proof = self.db.require_valid_assignment(
|
|
session_id="s1",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=7,
|
|
action="implement",
|
|
)
|
|
self.assertEqual(proof["session_id"], "s1")
|
|
with self.assertRaises(LeaseRequiredError):
|
|
self.db.require_valid_assignment(
|
|
session_id="s1",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=7,
|
|
action="merge",
|
|
)
|
|
with self.assertRaises(LeaseRequiredError):
|
|
self.db.require_valid_assignment(
|
|
session_id="s-other",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=7,
|
|
action="implement",
|
|
)
|
|
|
|
def test_expired_lease_allows_reassign(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
self.db.upsert_session(session_id="s2", role="author")
|
|
past = _utc_now() - timedelta(hours=1)
|
|
# Create lease already expired by using negative TTL edge via direct assign then expire
|
|
assigned = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=3,
|
|
lease_ttl_seconds=1,
|
|
)
|
|
self.assertEqual(assigned.outcome, "assigned")
|
|
# Force expiry in DB
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"UPDATE leases SET expires_at = ? WHERE lease_id = ?",
|
|
(_ts(past), assigned.lease_id),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
n = self.db.expire_stale_leases()
|
|
self.assertGreaterEqual(n, 1)
|
|
second = self.db.assign_and_lease(
|
|
session_id="s2",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=3,
|
|
)
|
|
self.assertEqual(second.outcome, "assigned")
|
|
self.assertEqual(second.session_id, "s2")
|
|
|
|
def test_merged_work_never_assigned(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="merger")
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=99,
|
|
state="merged",
|
|
)
|
|
result = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="merger",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=99,
|
|
expected_head_sha="merged-head",
|
|
)
|
|
self.assertEqual(result.outcome, "no_safe_work")
|
|
|
|
def test_four_concurrent_sessions_unique_assignments(self) -> None:
|
|
"""Four concurrent assigners on four different issues — all succeed uniquely.
|
|
|
|
Also two concurrent assigners on the *same* issue: at most one assigned.
|
|
"""
|
|
for i in range(4):
|
|
self.db.upsert_session(session_id=f"sess-{i}", role="author")
|
|
|
|
def claim_unique(i: int):
|
|
return self.db.assign_and_lease(
|
|
session_id=f"sess-{i}",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=1000 + i,
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = [f.result() for f in as_completed([pool.submit(claim_unique, i) for i in range(4)])]
|
|
self.assertEqual({r.outcome for r in results}, {"assigned"})
|
|
numbers = sorted(r.work_number for r in results)
|
|
self.assertEqual(numbers, [1000, 1001, 1002, 1003])
|
|
|
|
# Contention on one item
|
|
self.db.upsert_session(session_id="c1", role="author")
|
|
self.db.upsert_session(session_id="c2", role="author")
|
|
self.db.upsert_session(session_id="c3", role="author")
|
|
self.db.upsert_session(session_id="c4", role="author")
|
|
barrier = threading.Barrier(4)
|
|
outcomes: list[str] = []
|
|
lock = threading.Lock()
|
|
|
|
def contend(sid: str) -> None:
|
|
barrier.wait()
|
|
res = self.db.assign_and_lease(
|
|
session_id=sid,
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=7777,
|
|
)
|
|
with lock:
|
|
outcomes.append(res.outcome)
|
|
|
|
threads = [threading.Thread(target=contend, args=(f"c{i}",)) for i in range(1, 5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
self.assertEqual(outcomes.count("assigned"), 1)
|
|
self.assertEqual(outcomes.count("wait"), 3)
|
|
|
|
def test_terminal_lock_index(self) -> None:
|
|
self.db.set_terminal_lock(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
terminal_pr=332,
|
|
review_id="rev-1",
|
|
decision="approve",
|
|
)
|
|
row = self.db.get_active_terminal_lock(remote="prgs", org="o", repo="r")
|
|
self.assertIsNotNone(row)
|
|
assert row is not None
|
|
self.assertEqual(row["terminal_pr"], 332)
|
|
self.assertEqual(row["status"], "active")
|
|
|
|
def test_incident_links_not_work_items(self) -> None:
|
|
link = self.db.upsert_incident_link(
|
|
provider="sentry",
|
|
provider_base_url="https://sentry.prgs.cc",
|
|
provider_org="prgs",
|
|
provider_project="gitea-tools-mcp",
|
|
provider_issue_id="12345",
|
|
gitea_org="Scaled-Tech-Consulting",
|
|
gitea_repo="Gitea-Tools",
|
|
gitea_issue_number=9001,
|
|
fingerprint="fp-1",
|
|
)
|
|
self.assertEqual(link["gitea_issue_number"], 9001)
|
|
found = self.db.get_incident_link_for_gitea_issue(
|
|
gitea_org="Scaled-Tech-Consulting",
|
|
gitea_repo="Gitea-Tools",
|
|
gitea_issue_number=9001,
|
|
)
|
|
self.assertIsNotNone(found)
|
|
# Linking does not create a work_item of incident kind
|
|
with self.assertRaises(InvalidWorkKindError):
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
kind="sentry",
|
|
number=12345,
|
|
)
|
|
|
|
def test_heartbeat_and_release(self) -> None:
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
a = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=1,
|
|
)
|
|
hb = self.db.heartbeat_lease(a.lease_id, session_id="s1")
|
|
self.assertEqual(hb["lease_id"], a.lease_id)
|
|
self.db.release_lease(a.lease_id, session_id="s1")
|
|
# After release another session can claim
|
|
self.db.upsert_session(session_id="s2", role="author")
|
|
b = self.db.assign_and_lease(
|
|
session_id="s2",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=1,
|
|
)
|
|
self.assertEqual(b.outcome, "assigned")
|
|
self.assertEqual(b.session_id, "s2")
|
|
|
|
def test_require_valid_assignment_rejects_stale_head(self) -> None:
|
|
"""Assignment must not authorize mutations after work-item head drifts."""
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=42,
|
|
expected_head_sha="head-v1",
|
|
allowed_actions=("implement",),
|
|
)
|
|
# Head drifts after assignment
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=42,
|
|
current_head_sha="head-v2",
|
|
)
|
|
with self.assertRaises(LeaseRequiredError) as ctx:
|
|
self.db.require_valid_assignment(
|
|
session_id="s1",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=42,
|
|
action="implement",
|
|
)
|
|
self.assertIn("stale head", str(ctx.exception).lower())
|
|
|
|
def test_require_valid_assignment_rejects_terminal_state(self) -> None:
|
|
"""Assignment must not authorize mutations after work item is merged/closed."""
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=55,
|
|
expected_head_sha="abc",
|
|
allowed_actions=("implement",),
|
|
)
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=55,
|
|
state="merged",
|
|
current_head_sha="abc",
|
|
)
|
|
with self.assertRaises(LeaseRequiredError) as ctx:
|
|
self.db.require_valid_assignment(
|
|
session_id="s1",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=55,
|
|
action="implement",
|
|
)
|
|
self.assertIn("terminal", str(ctx.exception).lower())
|
|
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=56,
|
|
state="open",
|
|
)
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=56,
|
|
allowed_actions=("implement",),
|
|
)
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=56,
|
|
state="closed",
|
|
)
|
|
with self.assertRaises(LeaseRequiredError):
|
|
self.db.require_valid_assignment(
|
|
session_id="s1",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="issue",
|
|
number=56,
|
|
action="implement",
|
|
)
|
|
|
|
def test_pr_assignment_requires_expected_head_sha(self) -> None:
|
|
"""PR assign and mutation must fail closed without a head pin."""
|
|
self.db.upsert_session(session_id="s1", role="author")
|
|
with self.assertRaises(LeaseRequiredError) as ctx:
|
|
self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="author",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=88,
|
|
expected_head_sha=None,
|
|
allowed_actions=("implement",),
|
|
)
|
|
self.assertIn("expected_head_sha", str(ctx.exception))
|
|
|
|
# Legacy path: force an unpinned PR assignment into the DB, then
|
|
# populate head and prove mutation is still rejected.
|
|
import sqlite3
|
|
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=89,
|
|
current_head_sha=None,
|
|
)
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
wid = conn.execute(
|
|
"SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89"
|
|
).fetchone()[0]
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status)
|
|
VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active')
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO leases(
|
|
lease_id, work_item_id, session_id, role, phase,
|
|
expires_at, heartbeat_at, status
|
|
) VALUES (
|
|
'lease-legacy', ?, 'legacy', 'author', 'claimed',
|
|
'2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active'
|
|
)
|
|
""",
|
|
(wid,),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO assignments(
|
|
assignment_id, work_item_id, session_id, lease_id,
|
|
allowed_actions, forbidden_actions, expected_head_sha,
|
|
role, status, created_at
|
|
) VALUES (
|
|
'asn-legacy', ?, 'legacy', 'lease-legacy',
|
|
'["implement"]', '["merge"]', NULL,
|
|
'author', 'active', '2020-01-01T00:00:00Z'
|
|
)
|
|
""",
|
|
(wid,),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=89,
|
|
current_head_sha="populated-head",
|
|
)
|
|
with self.assertRaises(LeaseRequiredError) as ctx2:
|
|
self.db.require_valid_assignment(
|
|
session_id="legacy",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=89,
|
|
action="implement",
|
|
)
|
|
self.assertIn("expected_head_sha pin", str(ctx2.exception))
|
|
|
|
def test_migrate_duplicate_null_scope_incident_links(self) -> None:
|
|
"""Legacy NULL-scope duplicates must migrate without UNIQUE crash."""
|
|
import sqlite3
|
|
|
|
from control_plane_db import ControlPlaneDB, ControlPlaneError
|
|
|
|
# Build a v1-like table with nullable scope columns and insert dups
|
|
# that collapse under normalization, then open ControlPlaneDB on it.
|
|
path = os.path.join(self._tmp.name, "legacy_dups.sqlite3")
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
CREATE TABLE incident_links (
|
|
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
provider TEXT NOT NULL,
|
|
provider_base_url TEXT,
|
|
provider_org TEXT,
|
|
provider_project TEXT,
|
|
provider_issue_id TEXT NOT NULL,
|
|
provider_short_id TEXT,
|
|
provider_permalink TEXT,
|
|
fingerprint TEXT,
|
|
gitea_org TEXT NOT NULL,
|
|
gitea_repo TEXT NOT NULL,
|
|
gitea_issue_number INTEGER NOT NULL,
|
|
linked_pr_numbers TEXT,
|
|
first_seen TEXT,
|
|
last_seen TEXT,
|
|
event_count INTEGER,
|
|
status TEXT NOT NULL DEFAULT 'open',
|
|
release_resolved_at TEXT,
|
|
last_sync_at TEXT,
|
|
UNIQUE (
|
|
provider, provider_base_url, provider_org,
|
|
provider_project, provider_issue_id
|
|
)
|
|
);
|
|
INSERT INTO incident_links(
|
|
provider, provider_base_url, provider_org, provider_project,
|
|
provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status
|
|
) VALUES
|
|
('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'),
|
|
('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open');
|
|
"""
|
|
)
|
|
# SQLite allows two NULL-scope rows with same provider/issue under UNIQUE.
|
|
n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
|
|
self.assertEqual(n, 2)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
db = ControlPlaneDB(path)
|
|
conn2 = sqlite3.connect(path)
|
|
try:
|
|
n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
|
|
rows = conn2.execute(
|
|
"SELECT provider_base_url, provider_org, provider_project, gitea_issue_number "
|
|
"FROM incident_links"
|
|
).fetchall()
|
|
finally:
|
|
conn2.close()
|
|
self.assertEqual(n2, 1)
|
|
self.assertEqual(rows[0][0], "")
|
|
self.assertEqual(rows[0][1], "")
|
|
self.assertEqual(rows[0][2], "")
|
|
self.assertEqual(rows[0][3], 10)
|
|
# Touch to silence unused import in type checkers if needed
|
|
self.assertTrue(issubclass(ControlPlaneError, Exception))
|
|
del db
|
|
|
|
def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None:
|
|
"""Conflicting Gitea targets for the same provider key must fail closed."""
|
|
import sqlite3
|
|
from control_plane_db import ControlPlaneDB, ControlPlaneError
|
|
|
|
path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3")
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE incident_links (
|
|
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
provider TEXT NOT NULL,
|
|
provider_base_url TEXT,
|
|
provider_org TEXT,
|
|
provider_project TEXT,
|
|
provider_issue_id TEXT NOT NULL,
|
|
gitea_org TEXT NOT NULL,
|
|
gitea_repo TEXT NOT NULL,
|
|
gitea_issue_number INTEGER NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'open',
|
|
UNIQUE (
|
|
provider, provider_base_url, provider_org,
|
|
provider_project, provider_issue_id
|
|
)
|
|
);
|
|
INSERT INTO incident_links(
|
|
provider, provider_base_url, provider_org, provider_project,
|
|
provider_issue_id, gitea_org, gitea_repo, gitea_issue_number
|
|
) VALUES
|
|
('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1),
|
|
('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
with self.assertRaises(ControlPlaneError) as ctx:
|
|
ControlPlaneDB(path)
|
|
self.assertIn("conflicting", str(ctx.exception).lower())
|
|
|
|
def test_migrate_conflicting_observation_metadata_fails_closed(self) -> None:
|
|
"""Same provider key + same Gitea target but differing obs metadata must fail closed.
|
|
|
|
Regression for silent data loss: migration used to keep lowest link_id and
|
|
delete peers after comparing only Gitea targets (#619 RC3).
|
|
"""
|
|
import sqlite3
|
|
from control_plane_db import ControlPlaneDB, ControlPlaneError
|
|
|
|
path = os.path.join(self._tmp.name, "legacy_meta_conflict.sqlite3")
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
CREATE TABLE incident_links (
|
|
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
provider TEXT NOT NULL,
|
|
provider_base_url TEXT,
|
|
provider_org TEXT,
|
|
provider_project TEXT,
|
|
provider_issue_id TEXT NOT NULL,
|
|
provider_short_id TEXT,
|
|
provider_permalink TEXT,
|
|
fingerprint TEXT,
|
|
gitea_org TEXT NOT NULL,
|
|
gitea_repo TEXT NOT NULL,
|
|
gitea_issue_number INTEGER NOT NULL,
|
|
linked_pr_numbers TEXT,
|
|
first_seen TEXT,
|
|
last_seen TEXT,
|
|
event_count INTEGER,
|
|
status TEXT NOT NULL DEFAULT 'open',
|
|
release_resolved_at TEXT,
|
|
last_sync_at TEXT,
|
|
UNIQUE (
|
|
provider, provider_base_url, provider_org,
|
|
provider_project, provider_issue_id
|
|
)
|
|
);
|
|
INSERT INTO incident_links(
|
|
provider, provider_base_url, provider_org, provider_project,
|
|
provider_issue_id, provider_permalink, fingerprint,
|
|
gitea_org, gitea_repo, gitea_issue_number,
|
|
event_count, status, first_seen, last_seen
|
|
) VALUES
|
|
(
|
|
'sentry', NULL, NULL, NULL, 'dup-meta',
|
|
'https://sentry.example/issues/1', 'fingerprint-A',
|
|
'org', 'repo', 42,
|
|
1, 'open', '2026-01-01T00:00:00Z', '2026-01-01T01:00:00Z'
|
|
),
|
|
(
|
|
'sentry', NULL, NULL, NULL, 'dup-meta',
|
|
'https://sentry.example/issues/1', 'fingerprint-B',
|
|
'org', 'repo', 42,
|
|
99, 'resolved', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z'
|
|
);
|
|
"""
|
|
)
|
|
n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
|
|
self.assertEqual(n, 2)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
with self.assertRaises(ControlPlaneError) as ctx:
|
|
ControlPlaneDB(path)
|
|
msg = str(ctx.exception).lower()
|
|
self.assertIn("conflicting", msg)
|
|
self.assertIn("observation metadata", msg)
|
|
|
|
# Rows must still be present — migration must not delete before failing.
|
|
conn2 = sqlite3.connect(path)
|
|
try:
|
|
remaining = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
|
|
fps = {
|
|
r[0]
|
|
for r in conn2.execute(
|
|
"SELECT fingerprint FROM incident_links ORDER BY link_id"
|
|
).fetchall()
|
|
}
|
|
finally:
|
|
conn2.close()
|
|
self.assertEqual(remaining, 2)
|
|
self.assertEqual(fps, {"fingerprint-A", "fingerprint-B"})
|
|
|
|
def test_incident_links_minimal_upsert_is_canonical(self) -> None:
|
|
"""Repeated minimal upserts must update one row (NULL-safe uniqueness)."""
|
|
a = self.db.upsert_incident_link(
|
|
provider="sentry",
|
|
provider_issue_id="inc-1",
|
|
gitea_org="org",
|
|
gitea_repo="repo",
|
|
gitea_issue_number=1,
|
|
)
|
|
b = self.db.upsert_incident_link(
|
|
provider="sentry",
|
|
provider_issue_id="inc-1",
|
|
gitea_org="org",
|
|
gitea_repo="repo",
|
|
gitea_issue_number=2,
|
|
# omit optional scope fields again
|
|
)
|
|
c = self.db.upsert_incident_link(
|
|
provider="sentry",
|
|
provider_issue_id="inc-1",
|
|
gitea_org="org",
|
|
gitea_repo="repo",
|
|
gitea_issue_number=3,
|
|
provider_base_url=None,
|
|
provider_org="",
|
|
provider_project=" ",
|
|
)
|
|
self.assertEqual(a["link_id"], b["link_id"])
|
|
self.assertEqual(b["link_id"], c["link_id"])
|
|
self.assertEqual(c["gitea_issue_number"], 3)
|
|
self.assertEqual(c["provider_base_url"], "")
|
|
self.assertEqual(c["provider_org"], "")
|
|
self.assertEqual(c["provider_project"], "")
|
|
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?",
|
|
("inc-1",),
|
|
).fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
self.assertEqual(n, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|