feat(acms): design and implement ACMS index data model with file metadata, tags, and tier assignment #10788
@@ -0,0 +1,250 @@
|
||||
Feature: ACMS Index Data Model
|
||||
As a developer building the ACMS
|
||||
I want a well-defined index data model with file metadata, tags, and tier assignment
|
||||
So that all ACMS components have a consistent foundational schema
|
||||
|
||||
Background:
|
||||
Given I have a stored UTC datetime "2026-04-01T10:00:00+00:00" as "created_at"
|
||||
And I have a stored UTC datetime "2026-04-15T12:00:00+00:00" as "last_accessed"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier enum
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: Tier enum has hot, warm, and cold values
|
||||
When I inspect the ACMS Tier enum
|
||||
Then the ACMS Tier enum should have value "hot"
|
||||
And the ACMS Tier enum should have value "warm"
|
||||
And the ACMS Tier enum should have value "cold"
|
||||
|
||||
Scenario: Default tier on new IndexEntry is hot
|
||||
When I create an ACMS IndexEntry with path "/src/main.py" and size 1024
|
||||
Then the ACMS entry tier should be "hot"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IndexEntry fields
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: IndexEntry stores all required fields
|
||||
When I create an ACMS IndexEntry with:
|
||||
| field | value |
|
||||
| path | /project/src/main.py |
|
||||
| size | 2048 |
|
||||
| checksum | abc123def456 |
|
||||
Then the ACMS entry path should be "/project/src/main.py"
|
||||
And the ACMS entry size should be 2048
|
||||
And the ACMS entry checksum should be "abc123def456"
|
||||
And the ACMS entry created_at should equal "created_at"
|
||||
And the ACMS entry last_accessed should equal "last_accessed"
|
||||
|
||||
Scenario: IndexEntry tags default to empty list
|
||||
When I create an ACMS IndexEntry with path "/src/main.py" and size 512
|
||||
Then the ACMS entry tags should be empty
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tag operations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: Add a tag to an IndexEntry
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I add tag "core" to the ACMS entry
|
||||
Then the ACMS entry should have tag "core"
|
||||
|
||||
Scenario: Add duplicate tag is idempotent
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I add tag "core" to the ACMS entry
|
||||
And I add tag "core" to the ACMS entry
|
||||
Then the ACMS entry should have exactly 1 tag
|
||||
|
||||
Scenario: Remove a tag from an IndexEntry
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core,important"
|
||||
When I remove tag "core" from the ACMS entry
|
||||
Then the ACMS entry should not have tag "core"
|
||||
And the ACMS entry should have tag "important"
|
||||
|
||||
Scenario: Remove a non-existent tag is a no-op
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I remove tag "nonexistent" from the ACMS entry
|
||||
Then the ACMS entry tags should be empty
|
||||
|
||||
Scenario: has_tag returns True for present tag
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core"
|
||||
Then the ACMS entry should have tag "core"
|
||||
|
||||
Scenario: has_tag returns False for absent tag
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
Then the ACMS entry should not have tag "missing"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialisation — JSON
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: to_dict produces a serialisable dictionary
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 512 and tags "core,api"
|
||||
When I serialise the ACMS entry to a dict
|
||||
Then the ACMS dict should have key "path" with value "/src/main.py"
|
||||
And the ACMS dict should have key "size" with value 512
|
||||
And the ACMS dict should have key "tier" with value "hot"
|
||||
And the ACMS dict should have key "tags" containing "core"
|
||||
And the ACMS dict should have key "tags" containing "api"
|
||||
And the ACMS dict should have key "checksum"
|
||||
And the ACMS dict should have key "last_accessed"
|
||||
And the ACMS dict should have key "created_at"
|
||||
|
||||
Scenario: from_dict round-trips an IndexEntry
|
||||
Given I have an ACMS IndexEntry with path "/src/utils.py" and size 256 and tags "util"
|
||||
When I serialise the ACMS entry to a dict
|
||||
And I deserialise an ACMS IndexEntry from the dict
|
||||
Then the ACMS deserialised entry path should be "/src/utils.py"
|
||||
And the ACMS deserialised entry size should be 256
|
||||
And the ACMS deserialised entry should have tag "util"
|
||||
And the ACMS deserialised entry tier should be "hot"
|
||||
|
||||
Scenario: to_json produces a valid JSON string
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 1024
|
||||
When I serialise the ACMS entry to JSON
|
||||
Then the ACMS JSON string should be valid JSON
|
||||
And the ACMS JSON should contain path "/src/main.py"
|
||||
|
||||
Scenario: from_json round-trips an IndexEntry
|
||||
Given I have an ACMS IndexEntry with path "/src/app.py" and size 4096 and tags "app,entry"
|
||||
When I serialise the ACMS entry to JSON
|
||||
And I deserialise an ACMS IndexEntry from the JSON string
|
||||
Then the ACMS deserialised entry path should be "/src/app.py"
|
||||
And the ACMS deserialised entry size should be 4096
|
||||
And the ACMS deserialised entry should have tag "app"
|
||||
And the ACMS deserialised entry should have tag "entry"
|
||||
|
||||
Scenario: from_dict defaults tier to hot when tier key is missing
|
||||
Given I have a raw ACMS dict without a tier key for path "/src/main.py" and size 100
|
||||
When I deserialise an ACMS IndexEntry from the dict
|
||||
Then the ACMS deserialised entry tier should be "hot"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialisation — msgpack (optional)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: to_msgpack raises ImportError when msgpack is not installed
|
||||
Given msgpack package is not installed
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I attempt to serialise the ACMS entry to msgpack
|
||||
Then an ACMS ImportError should be raised
|
||||
|
||||
Scenario: from_msgpack raises ImportError when msgpack is not installed
|
||||
Given msgpack package is not installed
|
||||
When I attempt to deserialise an ACMS IndexEntry from msgpack bytes
|
||||
Then an ACMS ImportError should be raised
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TagIndex — O(1) lookup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Scenario: TagIndex lookup returns empty set for unknown tag
|
||||
Given I have an empty ACMS TagIndex
|
||||
When I look up ACMS tag "unknown" in the TagIndex
|
||||
Then the ACMS lookup result should be an empty set
|
||||
|
||||
Scenario: TagIndex add_entry enables tag lookup
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I look up ACMS tag "core" in the TagIndex
|
||||
Then the ACMS lookup result should contain path "/src/main.py"
|
||||
|
||||
Scenario: TagIndex lookup is O(1) — returns set of paths
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/a.py" and size 10 and tags "shared"
|
||||
And I have an ACMS IndexEntry with path "/src/b.py" and size 20 and tags "shared"
|
||||
When I add both ACMS entries to the TagIndex
|
||||
And I look up ACMS tag "shared" in the TagIndex
|
||||
Then the ACMS lookup result should contain path "/src/a.py"
|
||||
And the ACMS lookup result should contain path "/src/b.py"
|
||||
|
||||
Scenario: TagIndex remove_entry cleans up tag mappings
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I remove the ACMS entry "/src/main.py" from the TagIndex
|
||||
And I look up ACMS tag "core" in the TagIndex
|
||||
Then the ACMS lookup result should be an empty set
|
||||
|
||||
Scenario: TagIndex remove_entry returns False for unknown path
|
||||
Given I have an empty ACMS TagIndex
|
||||
When I remove the ACMS entry "/nonexistent.py" from the TagIndex
|
||||
Then the ACMS removal result should be False
|
||||
|
||||
Scenario: TagIndex remove_entry returns True for known path
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I remove the ACMS entry "/src/main.py" from the TagIndex
|
||||
Then the ACMS removal result should be True
|
||||
|
||||
Scenario: TagIndex add_tag updates lookup
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I add ACMS tag "new-tag" to path "/src/main.py" in the TagIndex
|
||||
And I look up ACMS tag "new-tag" in the TagIndex
|
||||
Then the ACMS lookup result should contain path "/src/main.py"
|
||||
|
||||
Scenario: TagIndex remove_tag updates lookup
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "core"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I remove ACMS tag "core" from path "/src/main.py" in the TagIndex
|
||||
And I look up ACMS tag "core" in the TagIndex
|
||||
Then the ACMS lookup result should be an empty set
|
||||
|
||||
Scenario: TagIndex get_entry returns the entry for a known path
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100
|
||||
When I add the ACMS entry to the TagIndex
|
||||
Then ACMS get_entry "/src/main.py" should return the entry
|
||||
|
||||
Scenario: TagIndex get_entry returns None for unknown path
|
||||
Given I have an empty ACMS TagIndex
|
||||
Then ACMS get_entry "/unknown.py" should return None
|
||||
|
||||
Scenario: TagIndex get_entries_by_tag returns matching entries
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/a.py" and size 10 and tags "api"
|
||||
And I have an ACMS IndexEntry with path "/src/b.py" and size 20 and tags "api,util"
|
||||
And I have an ACMS IndexEntry with path "/src/c.py" and size 30 and tags "util"
|
||||
When I add all three ACMS entries to the TagIndex
|
||||
And I get ACMS entries by tag "api" from the TagIndex
|
||||
Then I should get 2 ACMS entries
|
||||
And the ACMS entries should include path "/src/a.py"
|
||||
And the ACMS entries should include path "/src/b.py"
|
||||
|
||||
Scenario: TagIndex all_entries returns all entries
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/a.py" and size 10
|
||||
And I have an ACMS IndexEntry with path "/src/b.py" and size 20
|
||||
When I add both ACMS entries to the TagIndex
|
||||
And I get all ACMS entries from the TagIndex
|
||||
Then I should get 2 ACMS entries
|
||||
|
||||
Scenario: TagIndex all_tags returns all tags
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/a.py" and size 10 and tags "core,api"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I get all ACMS tags from the TagIndex
|
||||
Then the ACMS tags should contain "core"
|
||||
And the ACMS tags should contain "api"
|
||||
|
||||
Scenario: TagIndex len returns entry count
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/a.py" and size 10
|
||||
And I have an ACMS IndexEntry with path "/src/b.py" and size 20
|
||||
When I add both ACMS entries to the TagIndex
|
||||
Then the ACMS TagIndex length should be 2
|
||||
|
||||
Scenario: TagIndex replacing an entry updates tag mappings
|
||||
Given I have an empty ACMS TagIndex
|
||||
And I have an ACMS IndexEntry with path "/src/main.py" and size 100 and tags "old-tag"
|
||||
When I add the ACMS entry to the TagIndex
|
||||
Given I have an ACMS IndexEntry with path "/src/main.py" and size 200 and tags "new-tag"
|
||||
|
|
||||
When I add the ACMS entry to the TagIndex
|
||||
And I look up ACMS tag "old-tag" in the TagIndex
|
||||
Then the ACMS lookup result should be an empty set
|
||||
@@ -0,0 +1,502 @@
|
||||
"""Step definitions for ACMS Index Data Model feature (issue #9970)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
from behave import given, then, when # type: ignore[import-untyped]
|
||||
from behave.runner import Context # type: ignore[import-untyped]
|
||||
|
||||
from cleveragents.acms.data_model import IndexEntry, TagIndex, Tier
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Background / shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_entry(
|
||||
path: str,
|
||||
size: int,
|
||||
tags: list[str] | None = None,
|
||||
tier: Tier = Tier.HOT,
|
||||
checksum: str = "",
|
||||
created_at: datetime | None = None,
|
||||
last_accessed: datetime | None = None,
|
||||
) -> IndexEntry:
|
||||
now = datetime.now(tz=UTC)
|
||||
return IndexEntry(
|
||||
path=path,
|
||||
size=size,
|
||||
last_accessed=last_accessed or now,
|
||||
created_at=created_at or now,
|
||||
tags=list(tags or []),
|
||||
tier=tier,
|
||||
checksum=checksum,
|
||||
)
|
||||
|
||||
|
||||
@given('I have a stored UTC datetime "{iso}" as "{name}"')
|
||||
def step_store_datetime(context: Context, iso: str, name: str) -> None:
|
||||
if not hasattr(context, "acms_datetimes"):
|
||||
context.acms_datetimes = {}
|
||||
context.acms_datetimes[name] = datetime.fromisoformat(iso)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier enum
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I inspect the ACMS Tier enum")
|
||||
def step_inspect_tier(context: Context) -> None:
|
||||
context.acms_tier_values = {t.value for t in Tier}
|
||||
|
||||
|
||||
@then('the ACMS Tier enum should have value "{value}"')
|
||||
def step_tier_has_value(context: Context, value: str) -> None:
|
||||
assert value in context.acms_tier_values, (
|
||||
f"Tier enum missing value '{value}'; got {context.acms_tier_values}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IndexEntry creation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I create an ACMS IndexEntry with path "{path}" and size {size:d}')
|
||||
def step_create_entry_simple(context: Context, path: str, size: int) -> None:
|
||||
context.acms_entry = _make_entry(path=path, size=size)
|
||||
|
||||
|
||||
@when("I create an ACMS IndexEntry with:")
|
||||
def step_create_entry_table(context: Context) -> None:
|
||||
data: dict[str, str] = {}
|
||||
for row in context.table:
|
||||
data[row["field"]] = row["value"]
|
||||
|
||||
created_at = getattr(context, "acms_datetimes", {}).get(
|
||||
"created_at"
|
||||
) or datetime.now(tz=UTC)
|
||||
last_accessed = getattr(context, "acms_datetimes", {}).get(
|
||||
"last_accessed"
|
||||
) or datetime.now(tz=UTC)
|
||||
|
||||
context.acms_entry = IndexEntry(
|
||||
path=str(data.get("path", "/default.py")),
|
||||
size=int(data.get("size", "0")),
|
||||
last_accessed=last_accessed,
|
||||
created_at=created_at,
|
||||
tags=list(str(data.get("tags", "")).split(",")) if data.get("tags") else [],
|
||||
tier=Tier(str(data.get("tier", Tier.HOT))),
|
||||
checksum=str(data.get("checksum", "")),
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry path should be "{path}"')
|
||||
def step_entry_path(context: Context, path: str) -> None:
|
||||
assert context.acms_entry.path == path, (
|
||||
f"Expected path '{path}', got '{context.acms_entry.path}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS entry size should be {size:d}")
|
||||
def step_entry_size(context: Context, size: int) -> None:
|
||||
assert context.acms_entry.size == size, (
|
||||
f"Expected size {size}, got {context.acms_entry.size}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry checksum should be "{checksum}"')
|
||||
def step_entry_checksum(context: Context, checksum: str) -> None:
|
||||
assert context.acms_entry.checksum == checksum, (
|
||||
f"Expected checksum '{checksum}', got '{context.acms_entry.checksum}'"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry created_at should equal "{name}"')
|
||||
def step_entry_created_at(context: Context, name: str) -> None:
|
||||
expected = context.acms_datetimes[name]
|
||||
assert context.acms_entry.created_at == expected, (
|
||||
f"Expected created_at {expected}, got {context.acms_entry.created_at}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry last_accessed should equal "{name}"')
|
||||
def step_entry_last_accessed(context: Context, name: str) -> None:
|
||||
expected = context.acms_datetimes[name]
|
||||
assert context.acms_entry.last_accessed == expected, (
|
||||
f"Expected last_accessed {expected}, got {context.acms_entry.last_accessed}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry tier should be "{tier}"')
|
||||
def step_entry_tier(context: Context, tier: str) -> None:
|
||||
assert str(context.acms_entry.tier) == tier, (
|
||||
f"Expected tier '{tier}', got '{context.acms_entry.tier}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS entry tags should be empty")
|
||||
def step_entry_tags_empty(context: Context) -> None:
|
||||
assert context.acms_entry.tags == [], (
|
||||
f"Expected empty tags, got {context.acms_entry.tags}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tag operations on IndexEntry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('I have an ACMS IndexEntry with path "{path}" and size {size:d}')
|
||||
def step_given_entry_simple(context: Context, path: str, size: int) -> None:
|
||||
entry = _make_entry(path=path, size=size)
|
||||
context.acms_entry = entry
|
||||
if not hasattr(context, "acms_entries"):
|
||||
context.acms_entries = []
|
||||
# Add to entries list for multi-entry scenarios
|
||||
context.acms_entries.append(entry)
|
||||
|
||||
|
||||
@given(
|
||||
'I have an ACMS IndexEntry with path "{path}" and size {size:d} and tags "{tags_str}"'
|
||||
)
|
||||
def step_given_entry_with_tags(
|
||||
context: Context, path: str, size: int, tags_str: str
|
||||
) -> None:
|
||||
tags = [t.strip() for t in tags_str.split(",") if t.strip()]
|
||||
entry = _make_entry(path=path, size=size, tags=tags)
|
||||
context.acms_entry = entry
|
||||
if not hasattr(context, "acms_entries"):
|
||||
context.acms_entries = []
|
||||
context.acms_entries.append(entry)
|
||||
|
||||
|
||||
@when('I add tag "{tag}" to the ACMS entry')
|
||||
def step_add_tag(context: Context, tag: str) -> None:
|
||||
context.acms_entry.add_tag(tag)
|
||||
|
||||
|
||||
@when('I remove tag "{tag}" from the ACMS entry')
|
||||
def step_remove_tag(context: Context, tag: str) -> None:
|
||||
context.acms_entry.remove_tag(tag)
|
||||
|
||||
|
||||
@then('the ACMS entry should have tag "{tag}"')
|
||||
def step_entry_has_tag(context: Context, tag: str) -> None:
|
||||
assert context.acms_entry.has_tag(tag), (
|
||||
f"Entry does not have tag '{tag}'; tags={context.acms_entry.tags}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entry should not have tag "{tag}"')
|
||||
def step_entry_not_has_tag(context: Context, tag: str) -> None:
|
||||
assert not context.acms_entry.has_tag(tag), (
|
||||
f"Entry unexpectedly has tag '{tag}'; tags={context.acms_entry.tags}"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS entry should have exactly {count:d} tag")
|
||||
def step_entry_tag_count(context: Context, count: int) -> None:
|
||||
assert len(context.acms_entry.tags) == count, (
|
||||
f"Expected {count} tag(s), got {len(context.acms_entry.tags)}: "
|
||||
f"{context.acms_entry.tags}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialisation — dict / JSON
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I serialise the ACMS entry to a dict")
|
||||
def step_serialise_to_dict(context: Context) -> None:
|
||||
context.acms_serialised_dict = context.acms_entry.to_dict()
|
||||
|
||||
|
||||
@when("I deserialise an ACMS IndexEntry from the dict")
|
||||
def step_deserialise_from_dict(context: Context) -> None:
|
||||
context.acms_deserialised = IndexEntry.from_dict(context.acms_serialised_dict)
|
||||
|
||||
|
||||
@given('I have a raw ACMS dict without a tier key for path "{path}" and size {size:d}')
|
||||
def step_raw_dict_no_tier(context: Context, path: str, size: int) -> None:
|
||||
now = datetime.now(tz=UTC).isoformat()
|
||||
context.acms_serialised_dict = {
|
||||
"path": path,
|
||||
"size": size,
|
||||
"last_accessed": now,
|
||||
"created_at": now,
|
||||
"tags": [],
|
||||
"checksum": "",
|
||||
}
|
||||
|
||||
|
||||
@then('the ACMS dict should have key "{key}" with value "{value}"')
|
||||
def step_dict_key_str(context: Context, key: str, value: str) -> None:
|
||||
d = context.acms_serialised_dict
|
||||
assert key in d, f"Key '{key}' not in dict; keys={list(d.keys())}"
|
||||
assert str(d[key]) == value, f"Expected dict['{key}'] == '{value}', got '{d[key]}'"
|
||||
|
||||
|
||||
@then('the ACMS dict should have key "{key}" with value {value:d}')
|
||||
def step_dict_key_int(context: Context, key: str, value: int) -> None:
|
||||
d = context.acms_serialised_dict
|
||||
assert key in d, f"Key '{key}' not in dict; keys={list(d.keys())}"
|
||||
assert d[key] == value, f"Expected dict['{key}'] == {value}, got {d[key]}"
|
||||
|
||||
|
||||
@then('the ACMS dict should have key "{key}" containing "{item}"')
|
||||
def step_dict_key_contains(context: Context, key: str, item: str) -> None:
|
||||
d = context.acms_serialised_dict
|
||||
assert key in d, f"Key '{key}' not in dict; keys={list(d.keys())}"
|
||||
assert item in d[key], f"Expected '{item}' in dict['{key}'], got {d[key]}"
|
||||
|
||||
|
||||
@then('the ACMS dict should have key "{key}"')
|
||||
def step_dict_has_key(context: Context, key: str) -> None:
|
||||
assert key in context.acms_serialised_dict, (
|
||||
f"Key '{key}' not in dict; keys={list(context.acms_serialised_dict.keys())}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS deserialised entry path should be "{path}"')
|
||||
def step_deserialised_path(context: Context, path: str) -> None:
|
||||
assert context.acms_deserialised.path == path, (
|
||||
f"Expected path '{path}', got '{context.acms_deserialised.path}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS deserialised entry size should be {size:d}")
|
||||
def step_deserialised_size(context: Context, size: int) -> None:
|
||||
assert context.acms_deserialised.size == size, (
|
||||
f"Expected size {size}, got {context.acms_deserialised.size}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS deserialised entry should have tag "{tag}"')
|
||||
def step_deserialised_has_tag(context: Context, tag: str) -> None:
|
||||
assert context.acms_deserialised.has_tag(tag), (
|
||||
f"Deserialised entry does not have tag '{tag}'; "
|
||||
f"tags={context.acms_deserialised.tags}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS deserialised entry tier should be "{tier}"')
|
||||
def step_deserialised_tier(context: Context, tier: str) -> None:
|
||||
assert str(context.acms_deserialised.tier) == tier, (
|
||||
f"Expected tier '{tier}', got '{context.acms_deserialised.tier}'"
|
||||
)
|
||||
|
||||
|
||||
@when("I serialise the ACMS entry to JSON")
|
||||
def step_serialise_to_json(context: Context) -> None:
|
||||
context.acms_json_str = context.acms_entry.to_json()
|
||||
|
||||
|
||||
@when("I deserialise an ACMS IndexEntry from the JSON string")
|
||||
def step_deserialise_from_json(context: Context) -> None:
|
||||
context.acms_deserialised = IndexEntry.from_json(context.acms_json_str)
|
||||
|
||||
|
||||
@then("the ACMS JSON string should be valid JSON")
|
||||
def step_json_valid(context: Context) -> None:
|
||||
try:
|
||||
json.loads(context.acms_json_str)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise AssertionError(f"JSON string is not valid: {exc}") from exc
|
||||
|
||||
|
||||
@then('the ACMS JSON should contain path "{path}"')
|
||||
def step_json_contains_path(context: Context, path: str) -> None:
|
||||
data = json.loads(context.acms_json_str)
|
||||
assert data.get("path") == path, (
|
||||
f"Expected JSON path '{path}', got '{data.get('path')}'"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialisation — msgpack (optional)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("msgpack package is not installed")
|
||||
def step_msgpack_not_installed(context: Context) -> None:
|
||||
# Patch sys.modules so that 'import msgpack' raises ImportError
|
||||
context.acms_msgpack_patcher = patch.dict(sys.modules, {"msgpack": None})
|
||||
context.acms_msgpack_patcher.start()
|
||||
|
||||
|
||||
def _cleanup_msgpack_patcher(context: Context) -> None:
|
||||
patcher = getattr(context, "acms_msgpack_patcher", None)
|
||||
if patcher is not None:
|
||||
patcher.stop()
|
||||
context.acms_msgpack_patcher = None
|
||||
|
||||
|
||||
@when("I attempt to serialise the ACMS entry to msgpack")
|
||||
def step_attempt_to_msgpack(context: Context) -> None:
|
||||
context.acms_raised_exception = None
|
||||
try:
|
||||
context.acms_entry.to_msgpack()
|
||||
except ImportError as exc:
|
||||
context.acms_raised_exception = exc
|
||||
finally:
|
||||
_cleanup_msgpack_patcher(context)
|
||||
|
||||
|
||||
@when("I attempt to deserialise an ACMS IndexEntry from msgpack bytes")
|
||||
def step_attempt_from_msgpack(context: Context) -> None:
|
||||
context.acms_raised_exception = None
|
||||
try:
|
||||
IndexEntry.from_msgpack(b"")
|
||||
except ImportError as exc:
|
||||
context.acms_raised_exception = exc
|
||||
finally:
|
||||
_cleanup_msgpack_patcher(context)
|
||||
|
||||
|
||||
@then("an ACMS ImportError should be raised")
|
||||
def step_import_error_raised(context: Context) -> None:
|
||||
assert isinstance(context.acms_raised_exception, ImportError), (
|
||||
f"Expected ImportError, got {type(context.acms_raised_exception)}: "
|
||||
f"{context.acms_raised_exception}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TagIndex
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("I have an empty ACMS TagIndex")
|
||||
def step_empty_tag_index(context: Context) -> None:
|
||||
context.acms_tag_index = TagIndex()
|
||||
context.acms_removal_result = None
|
||||
context.acms_entries = []
|
||||
|
||||
|
||||
@when('I look up ACMS tag "{tag}" in the TagIndex')
|
||||
def step_lookup_tag(context: Context, tag: str) -> None:
|
||||
context.acms_lookup_result = context.acms_tag_index.lookup(tag)
|
||||
|
||||
|
||||
@then("the ACMS lookup result should be an empty set")
|
||||
def step_result_empty_set(context: Context) -> None:
|
||||
assert context.acms_lookup_result == set(), (
|
||||
f"Expected empty set, got {context.acms_lookup_result}"
|
||||
)
|
||||
|
||||
|
||||
@when("I add the ACMS entry to the TagIndex")
|
||||
def step_add_entry_to_tag_index(context: Context) -> None:
|
||||
context.acms_tag_index.add_entry(context.acms_entry)
|
||||
|
||||
|
||||
@then('the ACMS lookup result should contain path "{path}"')
|
||||
def step_result_contains_path(context: Context, path: str) -> None:
|
||||
assert path in context.acms_lookup_result, (
|
||||
f"Expected '{path}' in result, got {context.acms_lookup_result}"
|
||||
)
|
||||
|
||||
|
||||
@when("I add both ACMS entries to the TagIndex")
|
||||
def step_add_both_entries(context: Context) -> None:
|
||||
for entry in context.acms_entries:
|
||||
context.acms_tag_index.add_entry(entry)
|
||||
|
||||
|
||||
@when('I remove the ACMS entry "{path}" from the TagIndex')
|
||||
def step_remove_entry_from_tag_index(context: Context, path: str) -> None:
|
||||
context.acms_removal_result = context.acms_tag_index.remove_entry(path)
|
||||
|
||||
|
||||
@then("the ACMS removal result should be False")
|
||||
def step_removal_false(context: Context) -> None:
|
||||
assert context.acms_removal_result is False, (
|
||||
f"Expected removal result False, got {context.acms_removal_result}"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS removal result should be True")
|
||||
def step_removal_true(context: Context) -> None:
|
||||
assert context.acms_removal_result is True, (
|
||||
f"Expected removal result True, got {context.acms_removal_result}"
|
||||
)
|
||||
|
||||
|
||||
@when('I add ACMS tag "{tag}" to path "{path}" in the TagIndex')
|
||||
def step_tag_index_add_tag(context: Context, tag: str, path: str) -> None:
|
||||
context.acms_tag_index.add_tag(path, tag)
|
||||
|
||||
|
||||
@when('I remove ACMS tag "{tag}" from path "{path}" in the TagIndex')
|
||||
def step_tag_index_remove_tag(context: Context, tag: str, path: str) -> None:
|
||||
context.acms_tag_index.remove_tag(path, tag)
|
||||
|
||||
|
||||
@then('ACMS get_entry "{path}" should return the entry')
|
||||
def step_get_entry_returns(context: Context, path: str) -> None:
|
||||
result = context.acms_tag_index.get_entry(path)
|
||||
assert result is not None, f"get_entry('{path}') returned None"
|
||||
assert result.path == path, f"Expected path '{path}', got '{result.path}'"
|
||||
|
||||
|
||||
@then('ACMS get_entry "{path}" should return None')
|
||||
def step_get_entry_none(context: Context, path: str) -> None:
|
||||
result = context.acms_tag_index.get_entry(path)
|
||||
assert result is None, f"Expected None, got {result}"
|
||||
|
||||
|
||||
@when('I get ACMS entries by tag "{tag}" from the TagIndex')
|
||||
def step_get_entries_by_tag(context: Context, tag: str) -> None:
|
||||
context.acms_entries_result = context.acms_tag_index.get_entries_by_tag(tag)
|
||||
|
||||
|
||||
@when("I get all ACMS entries from the TagIndex")
|
||||
def step_get_all_entries(context: Context) -> None:
|
||||
context.acms_entries_result = context.acms_tag_index.all_entries()
|
||||
|
||||
|
||||
@when("I get all ACMS tags from the TagIndex")
|
||||
def step_get_all_tags(context: Context) -> None:
|
||||
context.acms_tags_result = context.acms_tag_index.all_tags()
|
||||
|
||||
|
||||
@then("I should get {count:d} ACMS entries")
|
||||
def step_should_get_n_entries(context: Context, count: int) -> None:
|
||||
assert len(context.acms_entries_result) == count, (
|
||||
f"Expected {count} entries, got {len(context.acms_entries_result)}"
|
||||
)
|
||||
|
||||
|
||||
@then('the ACMS entries should include path "{path}"')
|
||||
def step_entries_include_path(context: Context, path: str) -> None:
|
||||
paths = {e.path for e in context.acms_entries_result}
|
||||
assert path in paths, f"Expected '{path}' in entries, got {paths}"
|
||||
|
||||
|
||||
@then('the ACMS tags should contain "{tag}"')
|
||||
def step_tags_contain(context: Context, tag: str) -> None:
|
||||
assert tag in context.acms_tags_result, (
|
||||
f"Expected '{tag}' in tags, got {context.acms_tags_result}"
|
||||
)
|
||||
|
||||
|
||||
@then("the ACMS TagIndex length should be {count:d}")
|
||||
def step_tag_index_len(context: Context, count: int) -> None:
|
||||
assert len(context.acms_tag_index) == count, (
|
||||
f"Expected TagIndex length {count}, got {len(context.acms_tag_index)}"
|
||||
)
|
||||
|
||||
|
||||
@when("I add all three ACMS entries to the TagIndex")
|
||||
def step_add_all_three(context: Context) -> None:
|
||||
for entry in context.acms_entries:
|
||||
context.acms_tag_index.add_entry(entry)
|
||||
@@ -198,6 +198,12 @@ def step_check_query_result_count_singular(context, count):
|
||||
)
|
||||
|
||||
|
||||
@then("I should get {count:d} entries")
|
||||
def step_check_query_entries_count(context, count):
|
||||
"""Verify the query returned the expected number of entries."""
|
||||
assert len(context.query_results) == count
|
||||
|
||||
|
||||
@then('the results should include "{path}"')
|
||||
def step_check_result_includes_path(context, path):
|
||||
"""Verify the query results include a specific path."""
|
||||
|
||||
@@ -6,7 +6,8 @@ inheritance mechanism for resolving named detail levels across the
|
||||
ontology hierarchy (Layer 3 -> Layer 2 -> Layer 1 -> Layer 0).
|
||||
|
||||
Also provides the ACMS index data model and file traversal engine for
|
||||
indexing large projects.
|
||||
indexing large projects, and the foundational ACMS index data model
|
||||
(Tier enum, IndexEntry, TagIndex) from issue #9970.
|
||||
|
||||
Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
|
||||
"""
|
||||
@@ -14,6 +15,8 @@ Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
|
||||
from __future__ import annotations
|
||||
|
||||
from cleveragents.acms import uko as _uko
|
||||
from cleveragents.acms.data_model import IndexEntry as ACMSIndexEntry
|
||||
from cleveragents.acms.data_model import TagIndex, Tier
|
||||
from cleveragents.acms.index import (
|
||||
ACMSIndex,
|
||||
FileTraversalEngine,
|
||||
@@ -72,7 +75,7 @@ from cleveragents.acms.uko import (
|
||||
resolve_detail_level,
|
||||
)
|
||||
|
||||
# Combine exports from both uko and index modules
|
||||
# Combine exports from uko, index, and data_model modules
|
||||
_uko_exports = list(_uko.__all__)
|
||||
_index_exports = [
|
||||
"ACMSIndex",
|
||||
@@ -81,5 +84,10 @@ _index_exports = [
|
||||
"IndexEntry",
|
||||
"TierLevel",
|
||||
]
|
||||
_data_model_exports = [
|
||||
"ACMSIndexEntry",
|
||||
"TagIndex",
|
||||
"Tier",
|
||||
]
|
||||
|
||||
__all__: list[str] = _uko_exports + _index_exports
|
||||
__all__: list[str] = _uko_exports + _index_exports + _data_model_exports
|
||||
|
||||
@@ -0,0 +1,351 @@
|
||||
"""ACMS Index Data Model.
|
||||
HAL9001
commented
BLOCKER — CHANGELOG not updated: This file introduces new public API ( Add an entry under the appropriate section, e.g.: Automated by CleverAgents Bot **BLOCKER — CHANGELOG not updated:** This file introduces new public API (`Tier` enum, `IndexEntry` dataclass, `TagIndex` class) but no entry was added to `CHANGELOG.md`. Per CONTRIBUTING.md, every commit introducing new user-visible functionality must include a CHANGELOG update in the same commit.
Add an entry under the appropriate section, e.g.:
```
### Added
- ACMS index data model: `Tier` enum (hot/warm/cold), `IndexEntry` dataclass
with JSON/msgpack serialisation, and `TagIndex` for O(1) tag-based lookup
(issue #9970)
```
---
Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker
|
||||
|
||||
Provides the foundational data model for the Advanced Context Management
|
||||
System (ACMS): the ``Tier`` enum, the ``IndexEntry`` dataclass, and the
|
||||
``TagIndex`` structure for O(1) tag-based lookup.
|
||||
|
||||
Serialization to/from JSON and msgpack is supported for warm/cold tier
|
||||
persistence.
|
||||
|
||||
Based on issue #9970.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from enum import StrEnum
|
||||
from typing import Any
|
||||
|
||||
|
||||
class Tier(StrEnum):
|
||||
"""Storage tier assignment for ACMS index entries.
|
||||
|
||||
Attributes:
|
||||
HOT: Frequently accessed, kept in memory.
|
||||
WARM: Less frequently accessed, persisted to disk.
|
||||
COLD: Rarely accessed, archived storage.
|
||||
"""
|
||||
|
||||
HOT = "hot"
|
||||
WARM = "warm"
|
||||
COLD = "cold"
|
||||
|
||||
|
||||
def _now_utc() -> datetime:
|
||||
"""Return the current UTC datetime."""
|
||||
return datetime.now(tz=UTC)
|
||||
|
||||
|
||||
@dataclass
|
||||
class IndexEntry:
|
||||
"""A single entry in the ACMS index.
|
||||
|
||||
Represents a file's metadata, tags, and tier assignment within the
|
||||
Advanced Context Management System.
|
||||
|
||||
Attributes:
|
||||
path: Absolute or relative file path.
|
||||
size: File size in bytes.
|
||||
last_accessed: Timestamp of the last access.
|
||||
created_at: Timestamp when the entry was created.
|
||||
tags: List of arbitrary string tags for categorisation.
|
||||
tier: Storage tier assignment (hot/warm/cold).
|
||||
checksum: SHA-256 (or similar) hex digest of the file content.
|
||||
"""
|
||||
|
||||
path: str
|
||||
size: int
|
||||
last_accessed: datetime
|
||||
created_at: datetime
|
||||
tags: list[str] = field(default_factory=list)
|
||||
tier: Tier = Tier.HOT
|
||||
checksum: str = ""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Tag helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def add_tag(self, tag: str) -> None:
|
||||
"""Add *tag* to this entry (no-op if already present)."""
|
||||
if tag not in self.tags:
|
||||
self.tags.append(tag)
|
||||
|
||||
def remove_tag(self, tag: str) -> None:
|
||||
"""Remove *tag* from this entry (no-op if not present)."""
|
||||
if tag in self.tags:
|
||||
self.tags.remove(tag)
|
||||
|
||||
def has_tag(self, tag: str) -> bool:
|
||||
"""Return ``True`` if this entry has *tag*."""
|
||||
return tag in self.tags
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Serialisation helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialise this entry to a plain Python dictionary.
|
||||
|
||||
Datetime values are encoded as ISO-8601 strings so the result is
|
||||
directly JSON-serialisable.
|
||||
|
||||
Returns:
|
||||
A dictionary representation of this entry.
|
||||
"""
|
||||
return {
|
||||
"path": self.path,
|
||||
"size": self.size,
|
||||
"last_accessed": self.last_accessed.isoformat(),
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"tags": list(self.tags),
|
||||
"tier": str(self.tier),
|
||||
"checksum": self.checksum,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> IndexEntry:
|
||||
"""Deserialise an entry from a plain Python dictionary.
|
||||
|
||||
Args:
|
||||
data: Dictionary as produced by :meth:`to_dict`.
|
||||
|
||||
Returns:
|
||||
A new :class:`IndexEntry` instance.
|
||||
"""
|
||||
return cls(
|
||||
path=str(data["path"]),
|
||||
size=int(data["size"]),
|
||||
last_accessed=datetime.fromisoformat(str(data["last_accessed"])),
|
||||
created_at=datetime.fromisoformat(str(data["created_at"])),
|
||||
tags=list(data.get("tags", [])),
|
||||
tier=Tier(str(data.get("tier", Tier.HOT))),
|
||||
checksum=str(data.get("checksum", "")),
|
||||
)
|
||||
|
||||
def to_json(self) -> str:
|
||||
"""Serialise this entry to a JSON string.
|
||||
|
||||
Returns:
|
||||
A JSON-encoded string representation of this entry.
|
||||
"""
|
||||
return json.dumps(self.to_dict())
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, raw: str) -> IndexEntry:
|
||||
"""Deserialise an entry from a JSON string.
|
||||
|
||||
Args:
|
||||
raw: JSON string as produced by :meth:`to_json`.
|
||||
|
||||
Returns:
|
||||
A new :class:`IndexEntry` instance.
|
||||
"""
|
||||
return cls.from_dict(json.loads(raw))
|
||||
|
||||
def to_msgpack(self) -> bytes:
|
||||
"""Serialise this entry to msgpack bytes.
|
||||
|
||||
Requires the ``msgpack`` package to be installed.
|
||||
|
||||
Returns:
|
||||
A msgpack-encoded bytes representation of this entry.
|
||||
|
||||
Raises:
|
||||
ImportError: If the ``msgpack`` package is not installed.
|
||||
"""
|
||||
try:
|
||||
import msgpack # type: ignore[import-untyped]
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"msgpack is required for msgpack serialisation. "
|
||||
"Install it with: pip install msgpack"
|
||||
) from exc
|
||||
return msgpack.packb(self.to_dict(), use_bin_type=True) # type: ignore[no-any-return]
|
||||
|
||||
@classmethod
|
||||
def from_msgpack(cls, raw: bytes) -> IndexEntry:
|
||||
"""Deserialise an entry from msgpack bytes.
|
||||
|
||||
Requires the ``msgpack`` package to be installed.
|
||||
|
||||
Args:
|
||||
raw: msgpack bytes as produced by :meth:`to_msgpack`.
|
||||
|
||||
Returns:
|
||||
A new :class:`IndexEntry` instance.
|
||||
|
||||
Raises:
|
||||
ImportError: If the ``msgpack`` package is not installed.
|
||||
"""
|
||||
try:
|
||||
import msgpack # type: ignore[import-untyped]
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"msgpack is required for msgpack deserialisation. "
|
||||
"Install it with: pip install msgpack"
|
||||
) from exc
|
||||
data: dict[str, Any] = msgpack.unpackb(raw, raw=False)
|
||||
return cls.from_dict(data)
|
||||
|
||||
|
||||
class TagIndex:
|
||||
"""Inverted index for O(1) tag-based lookup of :class:`IndexEntry` objects.
|
||||
|
||||
Maintains a mapping from tag strings to the set of entry paths that
|
||||
carry that tag. All mutating operations keep the index consistent.
|
||||
|
||||
Example::
|
||||
|
||||
idx = TagIndex()
|
||||
entry = IndexEntry(path="/src/main.py", size=1024,
|
||||
last_accessed=datetime.now(tz=UTC),
|
||||
created_at=datetime.now(tz=UTC),
|
||||
tags=["core"])
|
||||
idx.add_entry(entry)
|
||||
paths = idx.lookup("core") # {"/src/main.py"}
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
# tag -> set of paths
|
||||
self._index: dict[str, set[str]] = {}
|
||||
# path -> entry (for retrieval)
|
||||
self._entries: dict[str, IndexEntry] = {}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Mutation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def add_entry(self, entry: IndexEntry) -> None:
|
||||
"""Add *entry* to the index.
|
||||
|
||||
If an entry with the same path already exists it is replaced.
|
||||
|
||||
Args:
|
||||
entry: The :class:`IndexEntry` to add.
|
||||
"""
|
||||
# Remove old tags if the entry already exists
|
||||
if entry.path in self._entries:
|
||||
self.remove_entry(entry.path)
|
||||
|
||||
self._entries[entry.path] = entry
|
||||
for tag in entry.tags:
|
||||
self._index.setdefault(tag, set()).add(entry.path)
|
||||
|
||||
def remove_entry(self, path: str) -> bool:
|
||||
"""Remove the entry identified by *path*.
|
||||
|
||||
Args:
|
||||
path: The file path of the entry to remove.
|
||||
|
||||
Returns:
|
||||
``True`` if the entry was found and removed, ``False`` otherwise.
|
||||
"""
|
||||
entry = self._entries.pop(path, None)
|
||||
if entry is None:
|
||||
return False
|
||||
for tag in entry.tags:
|
||||
paths = self._index.get(tag)
|
||||
if paths is not None:
|
||||
paths.discard(path)
|
||||
if not paths:
|
||||
del self._index[tag]
|
||||
return True
|
||||
|
||||
def add_tag(self, path: str, tag: str) -> None:
|
||||
"""Add *tag* to the entry identified by *path*.
|
||||
|
||||
Args:
|
||||
path: The file path of the entry.
|
||||
tag: The tag to add.
|
||||
|
||||
Raises:
|
||||
KeyError: If no entry with *path* exists in the index.
|
||||
"""
|
||||
entry = self._entries[path]
|
||||
entry.add_tag(tag)
|
||||
self._index.setdefault(tag, set()).add(path)
|
||||
|
||||
def remove_tag(self, path: str, tag: str) -> None:
|
||||
"""Remove *tag* from the entry identified by *path*.
|
||||
|
||||
Args:
|
||||
path: The file path of the entry.
|
||||
tag: The tag to remove.
|
||||
|
||||
Raises:
|
||||
KeyError: If no entry with *path* exists in the index.
|
||||
"""
|
||||
entry = self._entries[path]
|
||||
entry.remove_tag(tag)
|
||||
paths = self._index.get(tag)
|
||||
if paths is not None:
|
||||
paths.discard(path)
|
||||
if not paths:
|
||||
del self._index[tag]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Query
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def lookup(self, tag: str) -> set[str]:
|
||||
"""Return the set of paths that have *tag*.
|
||||
|
||||
Args:
|
||||
tag: The tag to look up.
|
||||
|
||||
Returns:
|
||||
A (possibly empty) set of file paths.
|
||||
"""
|
||||
return set(self._index.get(tag, set()))
|
||||
|
||||
def get_entry(self, path: str) -> IndexEntry | None:
|
||||
"""Return the entry for *path*, or ``None`` if not found.
|
||||
|
||||
Args:
|
||||
path: The file path to look up.
|
||||
|
||||
Returns:
|
||||
The :class:`IndexEntry` or ``None``.
|
||||
"""
|
||||
return self._entries.get(path)
|
||||
|
||||
def get_entries_by_tag(self, tag: str) -> list[IndexEntry]:
|
||||
"""Return all entries that have *tag*.
|
||||
|
||||
Args:
|
||||
tag: The tag to filter by.
|
||||
|
||||
Returns:
|
||||
A list of matching :class:`IndexEntry` objects.
|
||||
"""
|
||||
paths = self._index.get(tag, set())
|
||||
return [self._entries[p] for p in paths if p in self._entries]
|
||||
|
||||
def all_entries(self) -> list[IndexEntry]:
|
||||
"""Return all entries in the index.
|
||||
|
||||
Returns:
|
||||
A list of all :class:`IndexEntry` objects.
|
||||
"""
|
||||
return list(self._entries.values())
|
||||
|
||||
def all_tags(self) -> set[str]:
|
||||
"""Return all tags currently in the index.
|
||||
|
||||
Returns:
|
||||
A set of tag strings.
|
||||
"""
|
||||
return set(self._index.keys())
|
||||
|
||||
def __len__(self) -> int:
|
||||
"""Return the number of entries in the index."""
|
||||
return len(self._entries)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"IndexEntry",
|
||||
"TagIndex",
|
||||
"Tier",
|
||||
]
|
||||
Reference in New Issue
Block a user
BLOCKER — Non-standard Gherkin structure (Given after When): This scenario uses
GivenafterWhensteps, which resets the step state machine mid-scenario. While Behave accepts this syntactically, it is non-standard and can cause unexpected behaviour in parallel test runners — and may be contributing to theunit_testsCI failure.Restructure so all
Givensteps appear beforeWhensteps. One way to achieve this is to add both entries tocontext.acms_entriesin theGivenphase, then useWhen I add both ACMS entries to the TagIndextwice (replacing the entry on the second call). Alternatively, introduce a separate step definition for the replacement scenario that takes two entries explicitly.Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker