mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
ab1a71c86c
* feat: add PostgreSQL CI integration tests Add --storage-backend pytest option and shared storage_backend fixture in conftest.py that creates SQLiteBackend or PostgreSQLBackend based on the flag. Migrate 13 storage test files to use shared fixture instead of local SQLiteBackend fixtures. Add test-postgres CI job with PostgreSQL 17 service container that runs the full test suite against real PostgreSQL. * fix: use TRUNCATE CASCADE for PG cleanup, wrap in try/finally TRUNCATE is faster than per-table DELETE and resets autoincrement sequences. try/except ensures reset_storage() always runs even if cleanup fails due to a corrupted connection from a failing test. * fix: document _engine coupling in PG cleanup comment
57 lines
2.5 KiB
Python
57 lines
2.5 KiB
Python
"""Tests for skill resource storage operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
|
|
class TestDeleteSkillResourceByPath:
|
|
def test_delete_existing(self, storage):
|
|
skill_id = uuid.uuid4().hex
|
|
rid = uuid.uuid4().hex
|
|
storage.create_skill_resource(rid, skill_id, "scripts/a.sh", "#!/bin/bash")
|
|
assert storage.delete_skill_resource_by_path(skill_id, "scripts/a.sh") is True
|
|
assert storage.get_skill_resource(skill_id, "scripts/a.sh") is None
|
|
|
|
def test_delete_not_found(self, storage):
|
|
assert storage.delete_skill_resource_by_path("nonexistent", "scripts/a.sh") is False
|
|
|
|
def test_delete_wrong_path(self, storage):
|
|
skill_id = uuid.uuid4().hex
|
|
rid = uuid.uuid4().hex
|
|
storage.create_skill_resource(rid, skill_id, "scripts/a.sh", "content")
|
|
assert storage.delete_skill_resource_by_path(skill_id, "scripts/b.sh") is False
|
|
# Original still exists
|
|
assert storage.get_skill_resource(skill_id, "scripts/a.sh") is not None
|
|
|
|
def test_delete_only_target(self, storage):
|
|
"""Deleting one resource doesn't affect others for the same skill."""
|
|
skill_id = uuid.uuid4().hex
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "scripts/a.sh", "a")
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "scripts/b.sh", "b")
|
|
assert storage.delete_skill_resource_by_path(skill_id, "scripts/a.sh") is True
|
|
assert storage.get_skill_resource(skill_id, "scripts/b.sh") is not None
|
|
assert len(storage.list_skill_resources(skill_id)) == 1
|
|
|
|
|
|
class TestListSkillResources:
|
|
def test_ordering(self, storage):
|
|
skill_id = uuid.uuid4().hex
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "scripts/z.sh", "z")
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "assets/a.txt", "a")
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "references/m.md", "m")
|
|
rows = storage.list_skill_resources(skill_id)
|
|
paths = [r["path"] for r in rows]
|
|
assert paths == sorted(paths)
|
|
|
|
def test_empty(self, storage):
|
|
assert storage.list_skill_resources("nonexistent") == []
|
|
|
|
def test_size_from_content(self, storage):
|
|
skill_id = uuid.uuid4().hex
|
|
content = "x" * 500
|
|
storage.create_skill_resource(uuid.uuid4().hex, skill_id, "scripts/a.sh", content)
|
|
rows = storage.list_skill_resources(skill_id)
|
|
assert len(rows) == 1
|
|
assert len(rows[0]["content"]) == 500
|