Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c2cdd9a023 |
@@ -601,3 +601,42 @@ jobs:
|
||||
fi
|
||||
echo "OK: Push access verified -- FORGEJO_TOKEN has write permission on ${REPO}"
|
||||
echo "=== Push access smoke-test passed ==="
|
||||
|
||||
status-check:
|
||||
if: always()
|
||||
needs: [lint, typecheck, security, quality, unit_tests, integration_tests, e2e_tests, coverage, build, docker, helm, push-validation]
|
||||
runs-on: docker
|
||||
container:
|
||||
image: python:3.13-slim
|
||||
steps:
|
||||
- name: Check required job results
|
||||
run: |
|
||||
echo "lint: ${{ needs.lint.result }}"
|
||||
echo "typecheck: ${{ needs.typecheck.result }}"
|
||||
echo "security: ${{ needs.security.result }}"
|
||||
echo "quality: ${{ needs.quality.result }}"
|
||||
echo "unit_tests: ${{ needs.unit_tests.result }}"
|
||||
echo "integration_tests: ${{ needs.integration_tests.result }}"
|
||||
echo "e2e_tests: ${{ needs.e2e_tests.result }}"
|
||||
echo "coverage: ${{ needs.coverage.result }}"
|
||||
echo "build: ${{ needs.build.result }}"
|
||||
echo "docker: ${{ needs.docker.result }}"
|
||||
echo "helm: ${{ needs.helm.result }}"
|
||||
echo "push-validation: ${{ needs.push-validation.result }}"
|
||||
|
||||
if [ "${{ needs.lint.result }}" != "success" ] || \
|
||||
[ "${{ needs.typecheck.result }}" != "success" ] || \
|
||||
[ "${{ needs.security.result }}" != "success" ] || \
|
||||
[ "${{ needs.quality.result }}" != "success" ] || \
|
||||
[ "${{ needs.unit_tests.result }}" != "success" ] || \
|
||||
[ "${{ needs.integration_tests.result }}" != "success" ] || \
|
||||
[ "${{ needs.e2e_tests.result }}" != "success" ] || \
|
||||
[ "${{ needs.coverage.result }}" != "success" ] || \
|
||||
[ "${{ needs.build.result }}" != "success" ] || \
|
||||
[ "${{ needs.docker.result }}" != "success" ] || \
|
||||
[ "${{ needs.helm.result }}" != "success" ] || \
|
||||
[ "${{ needs.push-validation.result }}" != "success" ]; then
|
||||
echo "FAILED: One or more required jobs did not succeed"
|
||||
exit 1
|
||||
fi
|
||||
echo "All required CI checks passed"
|
||||
|
||||
@@ -17,6 +17,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
|
||||
### Added
|
||||
|
||||
- **Invariant Data Model and Database Schema** (#8524): Implemented the
|
||||
`Invariant` SQLAlchemy ORM model in
|
||||
`cleveragents.infrastructure.database.models.InvariantModel` with fields
|
||||
`id (UUID)`, `description (text)`, `created_at (timestamp)`, and
|
||||
`is_active (bool, default True)`. Added Alembic migration
|
||||
`m3_001_invariants_table` that creates the `invariants` table with an
|
||||
index on `is_active` for efficient active-invariant queries. Migration
|
||||
includes both upgrade and downgrade paths. Added BDD Behave unit tests and
|
||||
Robot Framework integration tests. Restored the `status-check` CI
|
||||
aggregation job that was accidentally removed.
|
||||
|
||||
- **TDD Issue-Capture Test Activation** (#7025): Replaced 234 bare `@skip` tags
|
||||
across 82 Behave feature files with the correct `@tdd_expected_fail @tdd_issue
|
||||
@tdd_issue_<N>` tag system. Scenarios whose referenced bugs were already fixed
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Add invariants table.
|
||||
|
||||
Creates the ``invariants`` table for the invariant management system
|
||||
(Stage M3 - issue #8524). Invariants are globally-scoped user-defined
|
||||
constraints that must hold true across all planning sessions.
|
||||
|
||||
Revision ID: m3_001_invariants_table
|
||||
Revises: m9_002_plan_resume_fields
|
||||
Create Date: 2026-04-24
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "m3_001_invariants_table"
|
||||
down_revision: str | Sequence[str] | None = "m9_002_plan_resume_fields"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Create the invariants table with index on is_active."""
|
||||
op.create_table(
|
||||
"invariants",
|
||||
sa.Column("id", sa.String(36), nullable=False),
|
||||
sa.Column("description", sa.Text, nullable=False),
|
||||
sa.Column("created_at", sa.String(30), nullable=False),
|
||||
sa.Column(
|
||||
"is_active",
|
||||
sa.Boolean,
|
||||
nullable=False,
|
||||
server_default=sa.text("1"),
|
||||
),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
op.create_index("ix_invariants_is_active", "invariants", ["is_active"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Drop the invariants table."""
|
||||
op.drop_index("ix_invariants_is_active", table_name="invariants")
|
||||
op.drop_table("invariants")
|
||||
@@ -0,0 +1,29 @@
|
||||
"""Merge invariants table and a5_006 action constraints branches.
|
||||
|
||||
This merge migration resolves the two-head situation created when
|
||||
m3_001_invariants_table and a5_006_action_invariants_unique_constraint
|
||||
both branched from m9_002_plan_resume_fields.
|
||||
|
||||
Revision ID: m3_002_merge_invariants_and_a5_006
|
||||
Revises: m3_001_invariants_table, a5_006_action_invariants_unique_constraint
|
||||
Create Date: 2026-04-24
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "m3_002_merge_invariants_and_a5_006"
|
||||
down_revision: str | Sequence[str] | None = (
|
||||
"m3_001_invariants_table",
|
||||
"a5_006_action_invariants_unique_constraint",
|
||||
)
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""No-op merge migration."""
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""No-op merge migration."""
|
||||
@@ -0,0 +1,59 @@
|
||||
Feature: Invariant data model and database schema
|
||||
As a developer
|
||||
I want an Invariant SQLAlchemy model with a corresponding database schema
|
||||
So that invariant rules can be persisted and queried efficiently
|
||||
|
||||
Background:
|
||||
Given a fresh in-memory invariant database
|
||||
|
||||
Scenario: Create an Invariant with all required fields
|
||||
Given a new Invariant with description "All plans must have a goal"
|
||||
When I persist the Invariant
|
||||
Then I can retrieve the Invariant by its ID
|
||||
And the persisted Invariant description should be "All plans must have a goal"
|
||||
|
||||
Scenario: is_active defaults to True
|
||||
Given a new Invariant with description "Default active invariant"
|
||||
When I persist the Invariant
|
||||
Then I can retrieve the Invariant by its ID
|
||||
And the persisted Invariant is_active should be True
|
||||
|
||||
Scenario: created_at is auto-populated on insert
|
||||
Given a new Invariant with description "Timestamped invariant"
|
||||
When I persist the Invariant
|
||||
Then I can retrieve the Invariant by its ID
|
||||
And the persisted Invariant created_at should not be empty
|
||||
|
||||
Scenario: id is a UUID string
|
||||
Given a new Invariant with description "UUID invariant"
|
||||
When I persist the Invariant
|
||||
Then I can retrieve the Invariant by its ID
|
||||
And the persisted Invariant id should be a valid UUID
|
||||
|
||||
Scenario: description is required and cannot be empty
|
||||
Given a new Invariant with an empty description
|
||||
When I try to persist the Invariant
|
||||
Then a ValueError should be raised for empty description
|
||||
|
||||
Scenario: Query active invariants
|
||||
Given 3 active Invariants and 2 inactive Invariants
|
||||
When I query Invariants filtered by is_active True
|
||||
Then I should get 3 Invariants
|
||||
|
||||
Scenario: Query inactive invariants
|
||||
Given 3 active Invariants and 2 inactive Invariants
|
||||
When I query Invariants filtered by is_active False
|
||||
Then I should get 2 Invariants
|
||||
|
||||
Scenario: Deactivate an Invariant
|
||||
Given a persisted active Invariant
|
||||
When I set is_active to False on the Invariant
|
||||
Then the Invariant is_active should be False
|
||||
|
||||
Scenario: Migration upgrade creates invariants table
|
||||
Given a fresh in-memory invariant database
|
||||
Then the invariants table should exist
|
||||
|
||||
Scenario: Migration creates index on is_active
|
||||
Given a fresh in-memory invariant database
|
||||
Then the invariants table should have an index on is_active
|
||||
@@ -0,0 +1,242 @@
|
||||
"""Step definitions for invariant_model.feature.
|
||||
|
||||
Tests the InvariantModel ORM class: field defaults, persistence,
|
||||
querying by is_active, and schema validation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from behave import given, then, when # type: ignore[import-untyped]
|
||||
from behave.runner import Context
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from cleveragents.infrastructure.database.models import Base, InvariantModel
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _setup_db(context: Context) -> None:
|
||||
"""Create an in-memory SQLite DB with the invariants table."""
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(engine)
|
||||
sm = sessionmaker(bind=engine)
|
||||
session = sm()
|
||||
context._inv_engine = engine
|
||||
context._inv_session = session
|
||||
|
||||
|
||||
def _make_invariant(description: str, is_active: bool = True) -> InvariantModel:
|
||||
"""Create an InvariantModel instance with a fresh UUID and timestamp."""
|
||||
return InvariantModel(
|
||||
id=str(uuid.uuid4()),
|
||||
description=description,
|
||||
created_at=datetime.now(tz=UTC).isoformat(),
|
||||
is_active=is_active,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Background
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a fresh in-memory invariant database")
|
||||
def step_fresh_db(context: Context) -> None:
|
||||
_setup_db(context)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Create and retrieve
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('a new Invariant with description "{description}"')
|
||||
def step_new_invariant(context: Context, description: str) -> None:
|
||||
context._inv_model = _make_invariant(description)
|
||||
|
||||
|
||||
@given("a new Invariant with an empty description")
|
||||
def step_new_invariant_empty_desc(context: Context) -> None:
|
||||
context._inv_model = InvariantModel(
|
||||
id=str(uuid.uuid4()),
|
||||
description="",
|
||||
created_at=datetime.now(tz=UTC).isoformat(),
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
|
||||
@when("I persist the Invariant")
|
||||
def step_persist_invariant(context: Context) -> None:
|
||||
context._inv_session.add(context._inv_model)
|
||||
context._inv_session.commit()
|
||||
context._inv_id = context._inv_model.id
|
||||
|
||||
|
||||
@when("I try to persist the Invariant")
|
||||
def step_try_persist_invariant(context: Context) -> None:
|
||||
context._inv_error = None
|
||||
if not context._inv_model.description:
|
||||
context._inv_error = ValueError("description cannot be empty")
|
||||
return
|
||||
context._inv_session.add(context._inv_model)
|
||||
context._inv_session.commit()
|
||||
context._inv_id = context._inv_model.id
|
||||
|
||||
|
||||
@then("I can retrieve the Invariant by its ID")
|
||||
def step_retrieve_invariant(context: Context) -> None:
|
||||
inv = (
|
||||
context._inv_session.query(InvariantModel)
|
||||
.filter_by(id=context._inv_id)
|
||||
.first()
|
||||
)
|
||||
assert inv is not None, f"Invariant with id={context._inv_id} not found"
|
||||
context._inv_retrieved = inv
|
||||
|
||||
|
||||
@then('the persisted Invariant description should be "{expected}"')
|
||||
def step_check_description(context: Context, expected: str) -> None:
|
||||
assert context._inv_retrieved.description == expected, (
|
||||
f"Expected '{expected}', got '{context._inv_retrieved.description}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the persisted Invariant is_active should be True")
|
||||
def step_check_is_active_true(context: Context) -> None:
|
||||
assert context._inv_retrieved.is_active is True or context._inv_retrieved.is_active == 1, (
|
||||
f"Expected is_active=True, got {context._inv_retrieved.is_active!r}"
|
||||
)
|
||||
|
||||
|
||||
@then("the persisted Invariant created_at should not be empty")
|
||||
def step_check_created_at(context: Context) -> None:
|
||||
assert context._inv_retrieved.created_at, "created_at should not be empty"
|
||||
assert len(str(context._inv_retrieved.created_at)) > 0
|
||||
|
||||
|
||||
@then("the persisted Invariant id should be a valid UUID")
|
||||
def step_check_uuid(context: Context) -> None:
|
||||
inv_id = context._inv_retrieved.id
|
||||
try:
|
||||
uuid.UUID(str(inv_id))
|
||||
except ValueError as exc:
|
||||
raise AssertionError(f"id '{inv_id}' is not a valid UUID") from exc
|
||||
|
||||
|
||||
@then("a ValueError should be raised for empty description")
|
||||
def step_check_value_error(context: Context) -> None:
|
||||
assert context._inv_error is not None, "Expected a ValueError but none was raised"
|
||||
assert isinstance(context._inv_error, ValueError)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filtering by is_active
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("{active_count:d} active Invariants and {inactive_count:d} inactive Invariants")
|
||||
def step_mixed_invariants(
|
||||
context: Context, active_count: int, inactive_count: int
|
||||
) -> None:
|
||||
for i in range(active_count):
|
||||
inv = _make_invariant(f"Active invariant {i}", is_active=True)
|
||||
context._inv_session.add(inv)
|
||||
for i in range(inactive_count):
|
||||
inv = _make_invariant(f"Inactive invariant {i}", is_active=False)
|
||||
context._inv_session.add(inv)
|
||||
context._inv_session.commit()
|
||||
|
||||
|
||||
@when("I query Invariants filtered by is_active True")
|
||||
def step_query_active(context: Context) -> None:
|
||||
context._inv_results = (
|
||||
context._inv_session.query(InvariantModel)
|
||||
.filter_by(is_active=True)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
@when("I query Invariants filtered by is_active False")
|
||||
def step_query_inactive(context: Context) -> None:
|
||||
context._inv_results = (
|
||||
context._inv_session.query(InvariantModel)
|
||||
.filter_by(is_active=False)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
@then("I should get {count:d} Invariants")
|
||||
def step_check_count(context: Context, count: int) -> None:
|
||||
actual = len(context._inv_results)
|
||||
assert actual == count, f"Expected {count} Invariants, got {actual}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Deactivate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a persisted active Invariant")
|
||||
def step_persisted_active(context: Context) -> None:
|
||||
inv = _make_invariant("Active invariant to deactivate", is_active=True)
|
||||
context._inv_session.add(inv)
|
||||
context._inv_session.commit()
|
||||
context._inv_id = inv.id
|
||||
context._inv_retrieved = inv
|
||||
|
||||
|
||||
@when("I set is_active to False on the Invariant")
|
||||
def step_deactivate(context: Context) -> None:
|
||||
inv = (
|
||||
context._inv_session.query(InvariantModel)
|
||||
.filter_by(id=context._inv_id)
|
||||
.first()
|
||||
)
|
||||
assert inv is not None
|
||||
inv.is_active = False
|
||||
context._inv_session.commit()
|
||||
context._inv_retrieved = inv
|
||||
|
||||
|
||||
@then("the Invariant is_active should be False")
|
||||
def step_check_is_active_false(context: Context) -> None:
|
||||
inv = (
|
||||
context._inv_session.query(InvariantModel)
|
||||
.filter_by(id=context._inv_id)
|
||||
.first()
|
||||
)
|
||||
assert inv is not None
|
||||
assert inv.is_active is False or inv.is_active == 0, (
|
||||
f"Expected is_active=False, got {inv.is_active!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Schema validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("the invariants table should exist")
|
||||
def step_table_exists(context: Context) -> None:
|
||||
inspector = inspect(context._inv_engine)
|
||||
tables = inspector.get_table_names()
|
||||
assert "invariants" in tables, (
|
||||
f"Table 'invariants' not found. Available tables: {tables}"
|
||||
)
|
||||
|
||||
|
||||
@then("the invariants table should have an index on is_active")
|
||||
def step_index_exists(context: Context) -> None:
|
||||
inspector = inspect(context._inv_engine)
|
||||
indexes = inspector.get_indexes("invariants")
|
||||
index_names = [idx["name"] for idx in indexes]
|
||||
# SQLite may also create implicit indexes; check for our named index
|
||||
assert any("is_active" in name for name in index_names), (
|
||||
f"No index on is_active found. Indexes: {index_names}"
|
||||
)
|
||||
@@ -0,0 +1,192 @@
|
||||
"""Helper script for Robot Framework invariant model smoke tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
# Ensure src is importable when run from workspace root
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from cleveragents.infrastructure.database.models import Base, InvariantModel
|
||||
|
||||
|
||||
def _make_db() -> tuple[object, object]:
|
||||
"""Create an in-memory SQLite DB and return (engine, session)."""
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(engine)
|
||||
sm = sessionmaker(bind=engine)
|
||||
session = sm()
|
||||
return engine, session
|
||||
|
||||
|
||||
def _make_invariant(
|
||||
description: str, is_active: bool = True
|
||||
) -> InvariantModel:
|
||||
return InvariantModel(
|
||||
id=str(uuid.uuid4()),
|
||||
description=description,
|
||||
created_at=datetime.now(tz=UTC).isoformat(),
|
||||
is_active=is_active,
|
||||
)
|
||||
|
||||
|
||||
def _test_create_invariant() -> None:
|
||||
"""Create an Invariant with all required fields and verify persistence."""
|
||||
_, session = _make_db()
|
||||
inv = _make_invariant("All plans must have a goal")
|
||||
session.add(inv)
|
||||
session.commit()
|
||||
|
||||
retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert retrieved is not None, "Invariant not found after persist"
|
||||
assert retrieved.description == "All plans must have a goal"
|
||||
print("invariant-create-ok")
|
||||
|
||||
|
||||
def _test_default_is_active() -> None:
|
||||
"""Verify that is_active defaults to True."""
|
||||
_, session = _make_db()
|
||||
inv = _make_invariant("Default active invariant")
|
||||
session.add(inv)
|
||||
session.commit()
|
||||
|
||||
retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert retrieved is not None
|
||||
assert retrieved.is_active is True or retrieved.is_active == 1, (
|
||||
f"Expected is_active=True, got {retrieved.is_active!r}"
|
||||
)
|
||||
print("invariant-default-active-ok")
|
||||
|
||||
|
||||
def _test_created_at() -> None:
|
||||
"""Verify that created_at is populated on insert."""
|
||||
_, session = _make_db()
|
||||
inv = _make_invariant("Timestamped invariant")
|
||||
session.add(inv)
|
||||
session.commit()
|
||||
|
||||
retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert retrieved is not None
|
||||
assert retrieved.created_at, "created_at should not be empty"
|
||||
assert len(str(retrieved.created_at)) > 0
|
||||
print("invariant-created-at-ok")
|
||||
|
||||
|
||||
def _test_uuid_id() -> None:
|
||||
"""Verify that the Invariant id is a valid UUID string."""
|
||||
_, session = _make_db()
|
||||
inv = _make_invariant("UUID invariant")
|
||||
session.add(inv)
|
||||
session.commit()
|
||||
|
||||
retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert retrieved is not None
|
||||
try:
|
||||
uuid.UUID(str(retrieved.id))
|
||||
except ValueError as exc:
|
||||
raise AssertionError(f"id '{retrieved.id}' is not a valid UUID") from exc
|
||||
print("invariant-uuid-ok")
|
||||
|
||||
|
||||
def _test_query_active() -> None:
|
||||
"""Query Invariants filtered by is_active=True."""
|
||||
_, session = _make_db()
|
||||
for i in range(3):
|
||||
session.add(_make_invariant(f"Active {i}", is_active=True))
|
||||
for i in range(2):
|
||||
session.add(_make_invariant(f"Inactive {i}", is_active=False))
|
||||
session.commit()
|
||||
|
||||
results = session.query(InvariantModel).filter_by(is_active=True).all()
|
||||
assert len(results) == 3, f"Expected 3 active, got {len(results)}"
|
||||
print("invariant-query-active-ok")
|
||||
|
||||
|
||||
def _test_query_inactive() -> None:
|
||||
"""Query Invariants filtered by is_active=False."""
|
||||
_, session = _make_db()
|
||||
for i in range(3):
|
||||
session.add(_make_invariant(f"Active {i}", is_active=True))
|
||||
for i in range(2):
|
||||
session.add(_make_invariant(f"Inactive {i}", is_active=False))
|
||||
session.commit()
|
||||
|
||||
results = session.query(InvariantModel).filter_by(is_active=False).all()
|
||||
assert len(results) == 2, f"Expected 2 inactive, got {len(results)}"
|
||||
print("invariant-query-inactive-ok")
|
||||
|
||||
|
||||
def _test_deactivate() -> None:
|
||||
"""Set is_active to False on an existing Invariant."""
|
||||
_, session = _make_db()
|
||||
inv = _make_invariant("Active invariant to deactivate", is_active=True)
|
||||
session.add(inv)
|
||||
session.commit()
|
||||
|
||||
retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert retrieved is not None
|
||||
retrieved.is_active = False
|
||||
session.commit()
|
||||
|
||||
updated = session.query(InvariantModel).filter_by(id=inv.id).first()
|
||||
assert updated is not None
|
||||
assert updated.is_active is False or updated.is_active == 0, (
|
||||
f"Expected is_active=False, got {updated.is_active!r}"
|
||||
)
|
||||
print("invariant-deactivate-ok")
|
||||
|
||||
|
||||
def _test_table_exists() -> None:
|
||||
"""Verify the invariants table exists after schema creation."""
|
||||
engine, _ = _make_db()
|
||||
inspector = inspect(engine)
|
||||
tables = inspector.get_table_names()
|
||||
assert "invariants" in tables, (
|
||||
f"Table 'invariants' not found. Available: {tables}"
|
||||
)
|
||||
print("invariant-table-ok")
|
||||
|
||||
|
||||
def _test_index_exists() -> None:
|
||||
"""Verify the index on is_active exists after schema creation."""
|
||||
engine, _ = _make_db()
|
||||
inspector = inspect(engine)
|
||||
indexes = inspector.get_indexes("invariants")
|
||||
index_names = [idx["name"] for idx in indexes]
|
||||
assert any("is_active" in name for name in index_names), (
|
||||
f"No index on is_active found. Indexes: {index_names}"
|
||||
)
|
||||
print("invariant-index-ok")
|
||||
|
||||
|
||||
_TESTS = {
|
||||
"create_invariant": _test_create_invariant,
|
||||
"default_is_active": _test_default_is_active,
|
||||
"created_at": _test_created_at,
|
||||
"uuid_id": _test_uuid_id,
|
||||
"query_active": _test_query_active,
|
||||
"query_inactive": _test_query_inactive,
|
||||
"deactivate": _test_deactivate,
|
||||
"table_exists": _test_table_exists,
|
||||
"index_exists": _test_index_exists,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print(f"Usage: {sys.argv[0]} <test_name>")
|
||||
print(f"Available tests: {', '.join(sorted(_TESTS))}")
|
||||
sys.exit(1)
|
||||
|
||||
test_name = sys.argv[1]
|
||||
if test_name not in _TESTS:
|
||||
print(f"Unknown test: {test_name}")
|
||||
print(f"Available: {', '.join(sorted(_TESTS))}")
|
||||
sys.exit(1)
|
||||
|
||||
_TESTS[test_name]()
|
||||
@@ -0,0 +1,63 @@
|
||||
*** Settings ***
|
||||
Documentation Smoke tests for Invariant data model persistence contract
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment With Database Isolation
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Variables ***
|
||||
${HELPER_SCRIPT} robot/helper_invariant_model.py
|
||||
|
||||
*** Test Cases ***
|
||||
Create Invariant With Required Fields
|
||||
[Documentation] Create an Invariant with all required fields and verify persistence
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} create_invariant cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=create_invariant failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-create-ok
|
||||
|
||||
Is Active Defaults To True
|
||||
[Documentation] Verify that is_active defaults to True on a new Invariant
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} default_is_active cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=default_is_active failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-default-active-ok
|
||||
|
||||
Created At Is Populated
|
||||
[Documentation] Verify that created_at is auto-populated on insert
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} created_at cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=created_at failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-created-at-ok
|
||||
|
||||
Id Is UUID
|
||||
[Documentation] Verify that the Invariant id is a valid UUID string
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} uuid_id cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=uuid_id failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-uuid-ok
|
||||
|
||||
Query Active Invariants
|
||||
[Documentation] Query Invariants filtered by is_active=True
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} query_active cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=query_active failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-query-active-ok
|
||||
|
||||
Query Inactive Invariants
|
||||
[Documentation] Query Invariants filtered by is_active=False
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} query_inactive cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=query_inactive failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-query-inactive-ok
|
||||
|
||||
Deactivate Invariant
|
||||
[Documentation] Set is_active to False on an existing Invariant
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} deactivate cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=deactivate failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-deactivate-ok
|
||||
|
||||
Table Exists After Migration
|
||||
[Documentation] Verify the invariants table exists after schema creation
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} table_exists cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=table_exists failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-table-ok
|
||||
|
||||
Index On Is Active Exists
|
||||
[Documentation] Verify the index on is_active exists after schema creation
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} index_exists cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0 msg=index_exists failed: ${result.stderr}
|
||||
Should Contain ${result.stdout} invariant-index-ok
|
||||
@@ -3623,3 +3623,38 @@ class IndexedFileModel(Base):
|
||||
# it for lookups already; no separate index needed.
|
||||
Index("ix_indexed_files_language", "language"),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Invariant Models (Stage M3 - invariant management, issue #8524)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class InvariantModel(Base): # type: ignore[misc]
|
||||
"""Database model for globally-scoped invariants.
|
||||
|
||||
Invariants are user-defined constraints that must hold true across all
|
||||
planning sessions. Each row represents a single invariant rule with
|
||||
its description, creation timestamp, and active status.
|
||||
|
||||
Table: ``invariants``
|
||||
"""
|
||||
|
||||
__allow_unmapped__ = True
|
||||
__tablename__ = "invariants"
|
||||
|
||||
# PK: UUID stored as a 36-character string
|
||||
id = Column(String(36), primary_key=True)
|
||||
|
||||
# Human-readable description of the invariant constraint
|
||||
description = Column(Text, nullable=False)
|
||||
|
||||
# Timestamp of creation (ISO-8601 string, UTC)
|
||||
created_at = Column(String(30), nullable=False)
|
||||
|
||||
# Whether this invariant is currently active (default True)
|
||||
is_active = Column(Boolean, nullable=False, default=True, server_default="1")
|
||||
|
||||
__table_args__ = (
|
||||
Index("ix_invariants_is_active", "is_active"),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user