cleveragents-core/robot/helper_domain_analyzers.py

#!/usr/bin/env python3
"""Robot Framework helper for domain analyzer integration tests."""

from __future__ import annotations

import sys
from collections.abc import Callable
from pathlib import Path

# Ensure worktree src is *first* on sys.path so it shadows any installed copy.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC in sys.path:
    sys.path.remove(_SRC)
sys.path.insert(0, _SRC)
# Clear cached cleveragents modules so our worktree src is used.
for _key in list(sys.modules):
    if _key.startswith("cleveragents"):
        del sys.modules[_key]

from cleveragents.domain.models.acms.analyzers import (  # noqa: E402
    AnalyzerProtocol,
    AnalyzerRegistry,
)
from cleveragents.domain.models.acms.docker_compose_analyzer import (  # noqa: E402
    DockerComposeAnalyzer,
)
from cleveragents.domain.models.acms.markdown_analyzer import (  # noqa: E402
    MarkdownAnalyzer,
)
from cleveragents.domain.models.acms.postgresql_analyzer import (  # noqa: E402
    PostgreSQLAnalyzer,
)
from cleveragents.domain.models.acms.python_analyzer import (  # noqa: E402
    PythonAnalyzer,
)

# ---------------------------------------------------------------------------
# Sample content
# ---------------------------------------------------------------------------

_SAMPLE_PYTHON = '''\
"""User authentication service for the application."""

from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class Credentials:
    """Holds user login credentials."""

    username: str
    password_hash: str
    is_active: bool = True
    roles: list[str] = field(default_factory=list)


class AuthenticationService:
    """Service responsible for authenticating users against the store."""

    def __init__(self, secret_key: str) -> None:
        self._secret_key = secret_key
        self._sessions: dict[str, Credentials] = {}

    def authenticate(self, username: str, password: str) -> Optional[str]:
        """Authenticate a user and return a session token.

        Args:
            username: The login name.
            password: The plaintext password to verify.

        Returns:
            A session token string on success, or None on failure.
        """
        pw_hash = hashlib.sha256(
            (password + self._secret_key).encode()
        ).hexdigest()
        logger.info("Authentication attempt for user=%s", username)
        return pw_hash if username else None

    def revoke_session(self, token: str) -> bool:
        """Revoke an active session by its token."""
        if token in self._sessions:
            del self._sessions[token]
            return True
        return False


def create_default_service() -> AuthenticationService:
    """Factory function to create a service with default settings."""
    return AuthenticationService(secret_key="default-dev-key")
'''

_SAMPLE_MARKDOWN = """\
# Architecture Overview

This document describes the high-level architecture of our platform.

## Core Components

The system is composed of several key services:

- **API Gateway** — routes external traffic
- **Auth Service** — handles user authentication
- **Data Pipeline** — processes incoming events

For more details see [the design doc](https://wiki.example.com/design).

### API Gateway

The gateway is built on [Envoy](https://envoyproxy.io) and handles
TLS termination, rate limiting, and request routing.

```yaml
listeners:
  - address: 0.0.0.0
    port: 8443
    tls: true
```

### Auth Service

Token-based authentication using JWT.  See [RFC 7519](https://tools.ietf.org/html/rfc7519).

```python
def verify_token(token: str) -> Claims:
    return jwt.decode(token, key=SECRET)
```

## Deployment

The platform runs on Kubernetes with Helm charts managed in the
`infra/` directory.  Refer to [ops runbook](./runbook.md) for details.
"""

_SAMPLE_POSTGRESQL = """\
CREATE SCHEMA IF NOT EXISTS app;

CREATE TABLE app.users (
    id          SERIAL PRIMARY KEY,
    username    VARCHAR(120) NOT NULL,
    email       VARCHAR(255) NOT NULL,
    password    VARCHAR(255) NOT NULL,
    is_active   BOOLEAN DEFAULT TRUE,
    created_at  TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE TABLE app.roles (
    id          SERIAL PRIMARY KEY,
    name        VARCHAR(80) NOT NULL,
    description TEXT
);

CREATE TABLE app.user_roles (
    user_id     INTEGER NOT NULL,
    role_id     INTEGER NOT NULL,
    granted_at  TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (user_id, role_id),
    FOREIGN KEY (user_id) REFERENCES app.users (id),
    FOREIGN KEY (role_id) REFERENCES app.roles (id)
);

CREATE OR REPLACE VIEW app.active_users AS
SELECT u.id, u.username, u.email
FROM app.users u
WHERE u.is_active = TRUE;
"""

_SAMPLE_DOCKER_COMPOSE = """\
services:
  web:
    image: myapp/web:latest
    ports:
      - "8080:80"
      - "8443:443"
    environment:
      DATABASE_URL: postgres://db:5432/app
      REDIS_URL: redis://cache:6379
      LOG_LEVEL: info
    depends_on:
      - db
      - cache
    volumes:
      - ./config:/app/config:ro

  db:
    image: postgres:16-alpine
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: app
      POSTGRES_USER: appuser
      POSTGRES_PASSWORD: secret
    volumes:
      - pgdata:/var/lib/postgresql/data

  cache:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  worker:
    image: myapp/worker:latest
    environment:
      DATABASE_URL: postgres://db:5432/app
      REDIS_URL: redis://cache:6379
    depends_on:
      - db
      - cache
"""

# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------


def _has_object_uri(triples: list, uri_fragment: str) -> bool:
    """Return True if any triple has *uri_fragment* in its object_uri."""
    return any(uri_fragment in t.object_uri for t in triples)


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------


def _cmd_analyze_python() -> int:
    """Create a PythonAnalyzer, analyze sample Python, verify triple types."""
    analyzer = PythonAnalyzer()
    triples = analyzer.analyze(_SAMPLE_PYTHON, "src/auth/service.py")

    if not triples:
        print("analyze-python-fail: no triples produced")
        return 1

    has_module = _has_object_uri(triples, "uko-code:Module")
    has_class = _has_object_uri(triples, "uko-py:Class")
    has_function = _has_object_uri(triples, "uko-py:Function")

    if not has_module:
        print("analyze-python-fail: missing Module type")
        return 1
    if not has_class:
        print("analyze-python-fail: missing Class type")
        return 1
    if not has_function:
        print("analyze-python-fail: missing Function type")
        return 1

    print(f"analyze-python-ok: {len(triples)} triples (Module+Class+Function found)")
    return 0


def _cmd_analyze_markdown() -> int:
    """Create a MarkdownAnalyzer, analyze sample markdown, verify triple types."""
    analyzer = MarkdownAnalyzer()
    triples = analyzer.analyze(_SAMPLE_MARKDOWN, "docs/architecture.md")

    if not triples:
        print("analyze-markdown-fail: no triples produced")
        return 1

    has_document = _has_object_uri(triples, "uko-doc:Document")
    has_section = _has_object_uri(triples, "uko-doc:Section")

    if not has_document:
        print("analyze-markdown-fail: missing Document type")
        return 1
    if not has_section:
        print("analyze-markdown-fail: missing Section type")
        return 1

    print(f"analyze-markdown-ok: {len(triples)} triples (Document+Section found)")
    return 0


def _cmd_analyze_postgresql() -> int:
    """Create a PostgreSQLAnalyzer, analyze sample DDL, verify triple types."""
    analyzer = PostgreSQLAnalyzer()
    triples = analyzer.analyze(_SAMPLE_POSTGRESQL, "db/migrations/001_init.sql")

    if not triples:
        print("analyze-postgresql-fail: no triples produced")
        return 1

    has_table = _has_object_uri(triples, "uko-data:Table")
    has_column = _has_object_uri(triples, "uko-data:Column")

    if not has_table:
        print("analyze-postgresql-fail: missing Table type")
        return 1
    if not has_column:
        print("analyze-postgresql-fail: missing Column type")
        return 1

    print(f"analyze-postgresql-ok: {len(triples)} triples (Table+Column found)")
    return 0


def _cmd_analyze_docker_compose() -> int:
    """Analyze sample compose YAML, verify triple types."""
    analyzer = DockerComposeAnalyzer()
    triples = analyzer.analyze(_SAMPLE_DOCKER_COMPOSE, "infra/docker-compose.yml")

    if not triples:
        print("analyze-docker-compose-fail: no triples produced")
        return 1

    has_service = _has_object_uri(triples, "uko-infra:Service")
    has_deployment = _has_object_uri(triples, "uko-infra:DeploymentUnit")

    if not has_service:
        print("analyze-docker-compose-fail: missing Service type")
        return 1
    if not has_deployment:
        print("analyze-docker-compose-fail: missing DeploymentUnit type")
        return 1

    print(
        f"analyze-docker-compose-ok: {len(triples)} triples"
        " (Service+DeploymentUnit found)"
    )
    return 0


def _cmd_protocol_check() -> int:
    """Verify all 4 analyzers satisfy AnalyzerProtocol."""
    analyzer_classes = [
        PythonAnalyzer,
        MarkdownAnalyzer,
        PostgreSQLAnalyzer,
        DockerComposeAnalyzer,
    ]

    for cls in analyzer_classes:
        inst = cls()
        if not isinstance(inst, AnalyzerProtocol):
            print(
                f"protocol-check-fail: {cls.__name__} doesn't satisfy AnalyzerProtocol"
            )
            return 1
        if not inst.domain:
            print(f"protocol-check-fail: {cls.__name__} has empty domain")
            return 1
        if not inst.supported_extensions:
            print(f"protocol-check-fail: {cls.__name__} has no supported extensions")
            return 1
        print(
            f"protocol-check: {cls.__name__} ok"
            f" (domain={inst.domain}, extensions={sorted(inst.supported_extensions)})"
        )

    print("protocol-check-ok")
    return 0


def _cmd_registry_check() -> int:
    """Create AnalyzerRegistry, register all 4 analyzers, verify lookup by extension."""
    registry = AnalyzerRegistry()
    analyzers = [
        PythonAnalyzer(),
        MarkdownAnalyzer(),
        PostgreSQLAnalyzer(),
        DockerComposeAnalyzer(),
    ]

    for analyzer in analyzers:
        registry.register(analyzer)

    if len(registry) != 4:
        print(f"registry-check-fail: expected 4 analyzers, got {len(registry)}")
        return 1

    # Verify extension lookup for each analyzer
    extension_checks = {
        ".py": "python",
        ".pyi": "python",
        ".md": "markdown",
        ".markdown": "markdown",
        ".sql": "postgresql",
        ".ddl": "postgresql",
        ".yml": "docker-compose",
        ".yaml": "docker-compose",
    }

    for ext, expected_domain in extension_checks.items():
        found = registry.get_for_extension(ext)
        if found is None:
            print(f"registry-check-fail: no analyzer found for '{ext}'")
            return 1
        if found.domain != expected_domain:
            print(
                f"registry-check-fail: expected domain '{expected_domain}'"
                f" for '{ext}', got '{found.domain}'"
            )
            return 1
        print(f"registry-check: '{ext}' -> {found.domain}")

    # Verify unknown extension returns None
    if registry.get_for_extension(".unknown") is not None:
        print("registry-check-fail: expected None for unknown extension")
        return 1

    print(f"registry-check-ok: {len(registry)} analyzers, all extensions verified")
    return 0


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], int]] = {
    "analyze-python": _cmd_analyze_python,
    "analyze-markdown": _cmd_analyze_markdown,
    "analyze-postgresql": _cmd_analyze_postgresql,
    "analyze-docker-compose": _cmd_analyze_docker_compose,
    "protocol-check": _cmd_protocol_check,
    "registry-check": _cmd_registry_check,
}


def main() -> int:
    """Run the specified command."""
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
        return 2
    return _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    sys.exit(main())