first commit

.dockerignore

@@ -0,0 +1,7 @@
+.git
+.venv
+__pycache__
+.pytest_cache
+.ruff_cache
+dist
+build

.env.example

@@ -0,0 +1,10 @@
+GITEA_BASE_URL=http://gitea.example.local
+GITEA_ADMIN_TOKEN=replace-me
+GITEA_DEFAULT_OWNER=simpl-open
+
+SMTP_HOST=smtp.example.local
+SMTP_PORT=587
+SMTP_USERNAME=
+SMTP_PASSWORD=
+SMTP_USE_TLS=true
+SMTP_FROM_ADDRESS=no-reply@example.local

.gitignore

@@ -0,0 +1,12 @@
+.venv/
+__pycache__/
+.pytest_cache/
+.pytest_cache
+.pytest
+.ruff_cache/
+.mypy_cache/
+dist/
+build/
+*.pyc
+*.egg-info/
+.dagster/

.pylintrc

@@ -0,0 +1,5 @@
+[MASTER]
+ignore=build,dist
+
+[MESSAGES CONTROL]
+disable=missing-module-docstring,missing-function-docstring,too-few-public-methods

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . ./
+
+ENV PYTHONPATH=/app/src
+
+CMD ["dagster", "dev", "-h", "0.0.0.0", "-p", "3000"]

README.md

@@ -0,0 +1,122 @@
+# Simpl-Open Credential Delivery
+
+A Dagster data service for generating Gitea credentials and delivering them to an application consumer by email.
+
+## Description
+
+This service extends the standard Dagster workflow template with a credential provisioning flow for Simpl-Open.
+
+The workflow:
+
+1. receives a repository identifier and consumer details
+2. generates Gitea credentials and repository access
+3. sends the generated credentials by email
+
+## Project Structure
+
+```
+simpl_open_credential_delivery/
+├── src/
+│   └── simpl_open_credential_delivery/
+│       ├── repository.py
+│       ├── config/
+│       ├── hooks/
+│       ├── jobs/
+│       ├── models/
+│       ├── ops/
+│       └── services/
+├── tests/
+├── documents/
+├── yaml/
+├── Dockerfile
+├── pyproject.toml
+├── requirements.txt
+└── workspace.yaml
+```
+
+## Getting Started
+
+### Prerequisites
+
+- Python 3.12+
+- uv or pip for dependency management
+
+### Installation
+
+```bash
+uv sync
+```
+
+Or with pip:
+
+```bash
+pip install -r requirements.txt
+```
+
+### Run Dagster
+
+```bash
+dagster dev
+```
+
+## Environment Variables
+
+### Gitea Integration
+
+- `GITEA_BASE_URL`: Base URL of the Gitea instance
+- `GITEA_ADMIN_TOKEN`: Admin token used to provision users, tokens, and repository access
+- `GITEA_DEFAULT_OWNER` (optional): Default repository owner used when the repository identifier is not fully qualified
+
+### Email Delivery
+
+- `SMTP_HOST`: SMTP server hostname
+- `SMTP_PORT`: SMTP server port
+- `SMTP_USERNAME` (optional): SMTP username
+- `SMTP_PASSWORD` (optional): SMTP password
+- `SMTP_USE_TLS` (default `true`): Enable STARTTLS
+- `SMTP_FROM_ADDRESS`: Sender email address
+
+## Workflow Input
+
+The workflow accepts the following high-level inputs:
+
+- `repository_identifier`
+- `consumer_email` with default literal `${CONSUMER_EMAIL}`
+- `access_level`
+
+Email subject and body templates are read from environment variables:
+
+- `EMAIL_SUBJECT_TEMPLATE`
+- `EMAIL_BODY_TEMPLATE`
+
+## Jobs
+
+### simpl_open_credentials_delivery_workflow
+
+Provision Gitea access credentials and send them by email to the consumer.
+
+## Development
+
+### Tests
+
+```bash
+pytest
+```
+
+### Docker Image
+
+```bash
+docker build -t simpl-open-credential-delivery:latest .
+```
+
+## License
+
+See LICENSE file for details.
+
+## Authors
+
+- Mat Basile
+
+## Version
+
+0.1.0

demo-config.yaml

@@ -0,0 +1,5 @@
+repository_identifier: simpl-open/platform-automation
+consumer_email: "${CONSUMER_EMAIL}"
+access_level: read-only
+token_name_prefix: simpl-open
+token_expiration_days: 30

documents/deployment-guide/deployment-guide.md

@@ -0,0 +1,5 @@
+# Deployment Guide
+
+The service can be deployed as a standard Dagster code location container.
+
+Set the environment variables for Gitea and SMTP before starting the container.

documents/installation-guide/installation-guide.md

@@ -0,0 +1,5 @@
+# Installation Guide
+
+1. Create and populate the required environment variables.
+2. Install dependencies with `uv sync` or `pip install -r requirements.txt`.
+3. Start Dagster with `dagster dev`.

documents/upgrade-guide/upgrade-guide.md

@@ -0,0 +1,3 @@
+# Upgrade Guide
+
+To extend the workflow, add new Dagster ops under `src/simpl_open_credential_delivery/ops` and wire them in the job module.

documents/user-manual/user-manual.md

@@ -0,0 +1,10 @@
+# User Manual
+
+Launch the `simpl_open_credentials_delivery_workflow` job with:
+
+- repository identifier
+- consumer identifier
+- consumer email
+- access level
+
+The workflow provisions credentials in Gitea and sends them by email.

pyproject.toml

@@ -0,0 +1,30 @@
+[project]
+name = "simpl_open_credential_delivery"
+version = "0.1.0"
+description = "Dagster code location for Simpl-Open credential delivery workflows"
+authors = [{ name = "Mat Basile" }]
+requires-python = ">=3.12"
+dependencies = [
+    "dagster>=1.11.16",
+    "dagster-webserver>=1.11.16",
+    "pytest>=8.4.2",
+    "pytest-cov>=7.0.0",
+    "requests>=2.32.0",
+]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+simpl_open_credential_delivery = ["*.py"]
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+testpaths = ["tests"]

requirements.txt

@@ -0,0 +1,5 @@
+dagster>=1.11.16
+dagster-webserver>=1.11.16
+pytest>=8.4.2
+pytest-cov>=7.0.0
+requests>=2.32.0

src/simpl_open_credential_delivery/__init__.py

@@ -0,0 +1,3 @@
+"""Simpl-Open credential delivery Dagster code location."""
+
+__version__ = "0.1.0"

src/simpl_open_credential_delivery/config/__init__.py

@@ -0,0 +1,6 @@
+"""Configuration helpers for Simpl-Open credential delivery workflows."""
+
+from simpl_open_credential_delivery.config.gitea_workflow_config_mapping import (
+    simpl_open_gitea_workflow_config_mapping,
+    simpl_open_gitea_workflow_config_mapping_fn,
+)

src/simpl_open_credential_delivery/config/gitea_workflow_config_mapping.py

@@ -0,0 +1,53 @@
+"""
+Configuration mapping for Simpl-Open credential delivery workflows.
+"""
+
+from dagster import Field, config_mapping
+
+
+def simpl_open_gitea_workflow_config_mapping_fn(cfg):
+    """Map high-level workflow configuration to op-level config."""
+    return {
+        "ops": {
+            "generate_gitea_credentials": {
+                "config": {
+                    "repository_identifier": cfg["repository_identifier"],
+                    "consumer_email": cfg["consumer_email"],
+                    "access_level": cfg["access_level"],
+                    "token_name_prefix": cfg["token_name_prefix"],
+                    "token_expiration_days": cfg["token_expiration_days"],
+                }
+            },
+            "send_credentials_email": {"config": {}},
+        }
+    }
+
+
+simpl_open_gitea_workflow_config_mapping = config_mapping(
+    config_schema={
+        "repository_identifier": Field(
+            str,
+            description="Gitea repository identifier in owner/repo format",
+        ),
+        "consumer_email": Field(
+            str,
+            default_value="${CONSUMER_EMAIL}",
+            description="Email address where credentials must be delivered",
+        ),
+        "access_level": Field(
+            str,
+            default_value="read-only",
+            description="Repository access level requested for the consumer",
+        ),
+        "token_name_prefix": Field(
+            str,
+            default_value="simpl-open",
+            description="Prefix used when creating the Gitea token name",
+        ),
+        "token_expiration_days": Field(
+            int,
+            default_value=30,
+            description="Token expiration in days",
+        ),
+    }
+)(simpl_open_gitea_workflow_config_mapping_fn)

src/simpl_open_credential_delivery/jobs/__init__.py

@@ -0,0 +1,5 @@
+"""Dagster jobs for Simpl-Open credential delivery."""
+
+from simpl_open_credential_delivery.jobs.simpl_open_credentials_delivery_workflow import (
+    simpl_open_credentials_delivery_workflow,
+)

src/simpl_open_credential_delivery/jobs/simpl_open_credentials_delivery_workflow.py

@@ -0,0 +1,29 @@
+"""
+Simpl-Open credential delivery workflow job.
+
+This job provisions Gitea credentials and sends them by email to the consumer.
+"""
+
+from dagster import job
+
+from simpl_open_credential_delivery.config import (
+    simpl_open_gitea_workflow_config_mapping,
+)
+from simpl_open_credential_delivery.ops import (
+    generate_gitea_credentials,
+    send_credentials_email,
+)
+
+
+@job(
+    name="simpl_open_credentials_delivery_workflow",
+    description=(
+        "Provision Gitea access credentials for a repository and deliver them by email"
+    ),
+    tags={"resource_type": "RD_APP"},
+    config=simpl_open_gitea_workflow_config_mapping,
+)
+def simpl_open_credentials_delivery_workflow():
+    """Main Simpl-Open credential delivery workflow."""
+    credentials = generate_gitea_credentials()
+    send_credentials_email(credentials)

src/simpl_open_credential_delivery/models/__init__.py

@@ -0,0 +1 @@
+"""Domain models for Simpl-Open credential delivery."""

src/simpl_open_credential_delivery/ops/__init__.py

@@ -0,0 +1,8 @@
+"""Dagster ops for Simpl-Open credential delivery."""
+
+from simpl_open_credential_delivery.ops.generate_gitea_credentials import (
+    generate_gitea_credentials,
+)
+from simpl_open_credential_delivery.ops.send_credentials_email import (
+    send_credentials_email,
+)

src/simpl_open_credential_delivery/ops/generate_gitea_credentials.py

@@ -0,0 +1,85 @@
+"""
+Generate Gitea credentials for a consumer request.
+"""
+
+import os
+
+from dagster import Field, OpExecutionContext, Out, op  # pylint: disable=import-error
+
+from simpl_open_credential_delivery.ops.retry_policy import DEFAULT_RETRY_POLICY
+from simpl_open_credential_delivery.services import GiteaAPIClient
+
+
+@op(
+    config_schema={
+        "repository_identifier": Field(
+            str,
+            description="Repository identifier in owner/repo format",
+        ),
+        "consumer_email": Field(
+            str,
+            description="Consumer email address",
+        ),
+        "access_level": Field(
+            str,
+            default_value="read-only",
+            description="Requested repository access level",
+        ),
+        "token_name_prefix": Field(
+            str,
+            default_value="simpl-open",
+            description="Prefix used for the generated access token name",
+        ),
+        "token_expiration_days": Field(
+            int,
+            default_value=30,
+            description="Token expiration in days",
+        ),
+    },
+    out=Out(dict, description="Generated credentials and repository URL"),
+    retry_policy=DEFAULT_RETRY_POLICY,
+    tags={"kind": "gitea_credentials"},
+)
+def generate_gitea_credentials(context: OpExecutionContext) -> dict:
+    """Provision Gitea credentials for the requested repository."""
+    config = context.op_config
+    base_url = os.getenv("GITEA_BASE_URL")
+    admin_token = os.getenv("GITEA_ADMIN_TOKEN") or os.getenv("GITEA_API_TOKEN")
+    default_owner = os.getenv("GITEA_DEFAULT_OWNER")
+
+    if not base_url:
+        raise ValueError("Missing required environment variable: GITEA_BASE_URL")
+    if not admin_token:
+        raise ValueError(
+            "Missing required environment variable: GITEA_ADMIN_TOKEN"
+        )
+
+    context.log.info("Generating Gitea credentials for repository %s", config["repository_identifier"])
+    client = GiteaAPIClient(
+        base_url=base_url,
+        admin_token=admin_token,
+        default_owner=default_owner,
+    )
+
+    credential_bundle = client.generate_credentials(
+        repository_identifier=config["repository_identifier"],
+        access_level=config["access_level"],
+        consumer_email=config["consumer_email"],
+        token_name_prefix=config["token_name_prefix"],
+        token_expiration_days=config["token_expiration_days"],
+    )
+
+    context.log.info(
+        "Generated credentials for %s with access level %s",
+        credential_bundle.username,
+        config["access_level"],
+    )
+
+    return {
+        "repository_identifier": config["repository_identifier"],
+        "consumer_email": config["consumer_email"],
+        "access_level": config["access_level"],
+        "username": credential_bundle.username,
+        "access_token": credential_bundle.access_token,
+        "repository_access_url": credential_bundle.repository_url,
+    }

src/simpl_open_credential_delivery/ops/retry_policy.py

@@ -0,0 +1,5 @@
+"""Shared retry policy for external service calls."""
+
+from dagster import RetryPolicy
+
+DEFAULT_RETRY_POLICY = RetryPolicy(max_retries=3, delay=5)

src/simpl_open_credential_delivery/ops/send_credentials_email.py

@@ -0,0 +1,43 @@
+"""
+Send generated credentials by email to the consumer.
+"""
+
+from dagster import Field, In, OpExecutionContext, Out, op  # pylint: disable=import-error
+
+from simpl_open_credential_delivery.ops.retry_policy import DEFAULT_RETRY_POLICY
+from simpl_open_credential_delivery.services import EmailDeliveryService
+
+
+@op(
+    ins={"credentials": In(dict)},
+    out=Out(dict, description="Email delivery result"),
+    retry_policy=DEFAULT_RETRY_POLICY,
+    tags={"kind": "email_delivery"},
+)
+def send_credentials_email(
+    context: OpExecutionContext,
+    credentials: dict,
+) -> dict:
+    """Send the generated credentials to the consumer email address."""
+    context.log.info(
+        "Sending credentials email to %s",
+        credentials["consumer_email"],
+    )
+
+    service = EmailDeliveryService.from_environment()
+    delivery_result = service.send_credentials_email(
+        recipient=credentials["consumer_email"],
+        credentials=credentials,
+    )
+
+    context.log.info(
+        "Credentials email sent to %s with message id %s",
+        delivery_result.recipient,
+        delivery_result.message_id,
+    )
+
+    return {
+        **credentials,
+        "email_delivery_status": "SENT",
+        "email_message_id": delivery_result.message_id,
+    }

src/simpl_open_credential_delivery/repository.py

@@ -0,0 +1,18 @@
+"""
+Repository definition for Simpl-Open credential delivery.
+
+This module exports the Dagster definitions for the credential provisioning
+workflow.
+"""
+
+from dagster import Definitions  # pylint: disable=import-error
+
+from simpl_open_credential_delivery.jobs import (
+    simpl_open_credentials_delivery_workflow,
+)
+
+defs = Definitions(
+    jobs=[
+        simpl_open_credentials_delivery_workflow,
+    ]
+)

src/simpl_open_credential_delivery/services/__init__.py

@@ -0,0 +1,8 @@
+"""External service clients for Simpl-Open credential delivery."""
+
+from simpl_open_credential_delivery.services.email_service import (
+    EmailDeliveryService,
+)
+from simpl_open_credential_delivery.services.gitea_api_client import (
+    GiteaAPIClient,
+)

src/simpl_open_credential_delivery/services/email_service.py

@@ -0,0 +1,106 @@
+"""
+Email delivery service used to send generated credentials.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+import smtplib
+from email.message import EmailMessage
+
+
+class EmailDeliveryError(RuntimeError):
+    """Raised when the email delivery step fails."""
+
+
+@dataclass(frozen=True)
+class EmailDeliveryResult:
+    """Result returned after email delivery."""
+
+    recipient: str
+    subject: str
+    message_id: str
+
+
+class EmailDeliveryService:
+    """SMTP based email delivery helper."""
+
+    def __init__(self, host: str, port: int, sender_address: str, subject_template: str, body_template: str, username: str | None = None, password: str | None = None, use_tls: bool = True, timeout_seconds: int = 30) -> None:
+        self.host = host
+        self.port = port
+        self.sender_address = sender_address
+        self.subject_template = subject_template
+        self.body_template = body_template
+        self.username = username
+        self.password = password
+        self.use_tls = use_tls
+        self.timeout_seconds = timeout_seconds
+
+    @classmethod
+    def from_environment(cls) -> "EmailDeliveryService":
+        """Create the service from environment variables."""
+        host = os.getenv("SMTP_HOST")
+        port = os.getenv("SMTP_PORT")
+        sender_address = os.getenv("SMTP_FROM_ADDRESS")
+
+        if not host:
+            raise EmailDeliveryError("Missing required environment variable: SMTP_HOST")
+        if not port:
+            raise EmailDeliveryError("Missing required environment variable: SMTP_PORT")
+        if not sender_address:
+            raise EmailDeliveryError("Missing required environment variable: SMTP_FROM_ADDRESS")
+
+        return cls(
+            host=host,
+            port=int(port),
+            sender_address=sender_address,
+            subject_template=os.getenv(
+                "EMAIL_SUBJECT_TEMPLATE",
+                "Simpl-Open access credentials for {repository_identifier}",
+            ),
+            body_template=os.getenv(
+                "EMAIL_BODY_TEMPLATE",
+                (
+                    "Hello {consumer_email},\n\n"
+                    "Your access credentials for {repository_identifier} are ready.\n\n"
+                    "Username: {username}\n"
+                    "Token: {access_token}\n"
+                    "Repository URL: {repository_access_url}\n"
+                    "Access level: {access_level}\n\n"
+                    "Regards,\n"
+                    "Simpl-Open Platform"
+                ),
+            ),
+            username=os.getenv("SMTP_USERNAME") or None,
+            password=os.getenv("SMTP_PASSWORD") or None,
+            use_tls=os.getenv("SMTP_USE_TLS", "true").lower() in {"1", "true", "yes", "on"},
+            timeout_seconds=int(os.getenv("SMTP_TIMEOUT_SECONDS", "30")),
+        )
+
+    def send_credentials_email(self, recipient: str, credentials: dict[str, str]) -> EmailDeliveryResult:
+        """Render the email from templates and send it using SMTP."""
+        subject = self.subject_template.format(**credentials)
+        body = self.body_template.format(**credentials)
+
+        message = EmailMessage()
+        message["To"] = recipient
+        message["From"] = self.sender_address
+        message["Subject"] = subject
+        message.set_content(body)
+
+        try:
+            with smtplib.SMTP(self.host, self.port, timeout=self.timeout_seconds) as smtp:
+                if self.use_tls:
+                    smtp.starttls()
+                if self.username and self.password:
+                    smtp.login(self.username, self.password)
+                smtp.send_message(message)
+        except Exception as exc:  # pylint: disable=broad-except
+            raise EmailDeliveryError(f"Failed to send credentials email: {exc}") from exc
+
+        return EmailDeliveryResult(
+            recipient=recipient,
+            subject=subject,
+            message_id=message.get("Message-ID", ""),
+        )

src/simpl_open_credential_delivery/services/gitea_api_client.py

@@ -0,0 +1,124 @@
+"""
+Official Gitea API client used by the credential provisioning workflow.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from typing import Any
+
+import requests
+
+
+class GiteaAPIError(RuntimeError):
+    """Raised when a Gitea API call fails."""
+
+
+@dataclass(frozen=True)
+class GiteaCredentialBundle:
+    """Credentials returned by the Gitea provisioning flow."""
+
+    username: str
+    access_token: str
+    repository_url: str
+
+
+class GiteaAPIClient:
+    """Minimal client for Gitea provisioning tasks."""
+
+    def __init__(self, base_url: str, admin_token: str, default_owner: str | None = None, timeout_seconds: int = 30) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.admin_token = admin_token
+        self.default_owner = default_owner
+        self.timeout_seconds = timeout_seconds
+
+    @staticmethod
+    def _slugify(value: str) -> str:
+        cleaned = re.sub(r"[^a-zA-Z0-9]+", "-", value.strip().lower())
+        return cleaned.strip("-") or "consumer"
+
+    def _request(self, method: str, path: str, expected_status: tuple[int, ...] = (200, 201, 204), **kwargs: Any) -> requests.Response:
+        response = requests.request(
+            method=method,
+            url=f"{self.base_url}{path}",
+            headers={
+                "Authorization": f"token {self.admin_token}",
+                "Accept": "application/json",
+            },
+            timeout=self.timeout_seconds,
+            **kwargs,
+        )
+        if response.status_code not in expected_status:
+            raise GiteaAPIError(
+                f"Gitea API call to {path} failed with status {response.status_code}: {response.text}"
+            )
+        return response
+
+    def _resolve_repository_identifier(self, repository_identifier: str) -> str:
+        if "/" in repository_identifier:
+            return repository_identifier
+        if self.default_owner:
+            return f"{self.default_owner}/{repository_identifier}"
+        raise ValueError(
+            "Repository identifier must be in owner/repo format or GITEA_DEFAULT_OWNER must be set"
+        )
+
+    @staticmethod
+    def _extract_token(payload: dict[str, Any]) -> str:
+        for key in ("token", "access_token", "sha1", "value"):
+            if key in payload and payload[key]:
+                return str(payload[key])
+        raise GiteaAPIError("Gitea token response did not contain a token value")
+
+    def generate_credentials(
+        self,
+        repository_identifier: str,
+        consumer_email: str,
+        access_level: str,
+        token_name_prefix: str,
+        token_expiration_days: int,
+    ) -> GiteaCredentialBundle:
+        """Provision a consumer user, access token, and repository access."""
+        resolved_repository = self._resolve_repository_identifier(repository_identifier)
+        owner, repository_name = resolved_repository.split("/", 1)
+        username_source = consumer_email.split("@", 1)[0] if "@" in consumer_email else consumer_email
+        username = self._slugify(username_source)
+        token_name = f"{token_name_prefix}-{self._slugify(repository_name)}-{username}"
+
+        self._request(
+            "POST",
+            "/api/v1/admin/users",
+            expected_status=(201, 409),
+            json={
+                "username": username,
+                "email": consumer_email,
+                "login_name": username,
+                "password": f"{username}-temporary-password",
+                "must_change_password": True,
+            },
+        )
+
+        token_response = self._request(
+            "POST",
+            f"/api/v1/users/{username}/tokens",
+            json={
+                "name": token_name,
+                "expires_in_days": token_expiration_days,
+            },
+        )
+        token_payload = token_response.json() if token_response.content else {}
+        access_token = self._extract_token(token_payload)
+
+        self._request(
+            "POST",
+            f"/api/v1/repos/{owner}/{repository_name}/collaborators/{username}",
+            expected_status=(201, 204),
+            json={"permission": access_level},
+        )
+
+        return GiteaCredentialBundle(
+            username=username,
+            access_token=access_token,
+            repository_url=f"{self.base_url}/{resolved_repository}",
+        )

tests/__init__.py

@@ -0,0 +1 @@
+"""Test package for Simpl-Open credential delivery."""

tests/conftest.py

@@ -0,0 +1,10 @@
+"""Pytest configuration for Simpl-Open credential delivery tests."""
+
+import pytest
+from dagster import build_op_context
+
+
+@pytest.fixture
+def op_context():
+    """Fixture for creating a basic op context."""
+    return build_op_context()

tests/test_config_mapping.py

@@ -0,0 +1,21 @@
+"""Tests for workflow config mapping."""
+
+from simpl_open_credential_delivery.config import (
+    simpl_open_gitea_workflow_config_mapping_fn,
+)
+
+
+def test_config_mapping_routes_inputs_to_ops():
+    cfg = simpl_open_gitea_workflow_config_mapping_fn(
+        {
+            "repository_identifier": "simpl-open/platform-automation",
+            "consumer_email": "consumer@example.local",
+            "access_level": "read-only",
+            "token_name_prefix": "simpl-open",
+            "token_expiration_days": 30,
+        }
+    )
+
+    assert cfg["ops"]["generate_gitea_credentials"]["config"]["repository_identifier"] == "simpl-open/platform-automation"
+    assert cfg["ops"]["generate_gitea_credentials"]["config"]["consumer_email"] == "consumer@example.local"
+    assert cfg["ops"]["send_credentials_email"]["config"] == {}

tests/test_generate_gitea_credentials.py

@@ -0,0 +1,70 @@
+"""Tests for Gitea credential generation."""
+
+import pytest
+
+from dagster import build_op_context
+
+from simpl_open_credential_delivery.ops.generate_gitea_credentials import (
+    generate_gitea_credentials,
+)
+
+
+class FakeResponse:
+    def __init__(self, status_code, payload=None, text="OK"):
+        self.status_code = status_code
+        self._payload = payload or {}
+        self.text = text
+        self.content = b"{}" if payload is not None else b""
+
+    def json(self):
+        return self._payload
+
+
+def test_generate_gitea_credentials_success(monkeypatch, op_context):
+    responses = [
+        FakeResponse(201),
+        FakeResponse(201, {"sha1": "token-123"}),
+        FakeResponse(204),
+    ]
+
+    def fake_request(*args, **kwargs):
+        return responses.pop(0)
+
+    monkeypatch.setenv("GITEA_BASE_URL", "https://gitea.example.local")
+    monkeypatch.setenv("GITEA_ADMIN_TOKEN", "admin-token")
+    monkeypatch.setenv("GITEA_DEFAULT_OWNER", "simpl-open")
+    monkeypatch.setattr("requests.request", fake_request)
+
+    op_context = build_op_context(
+        op_config={
+        "repository_identifier": "platform-automation",
+        "consumer_email": "consumer@example.local",
+        "access_level": "read-only",
+        "token_name_prefix": "simpl-open",
+        "token_expiration_days": 30,
+        }
+    )
+
+    result = generate_gitea_credentials(op_context)
+
+    assert result["username"] == "consumer"
+    assert result["access_token"] == "token-123"
+    assert result["repository_access_url"] == "https://gitea.example.local/simpl-open/platform-automation"
+
+
+def test_generate_gitea_credentials_requires_gitea_base_url(monkeypatch, op_context):
+    monkeypatch.delenv("GITEA_BASE_URL", raising=False)
+    monkeypatch.setenv("GITEA_ADMIN_TOKEN", "admin-token")
+
+    op_context = build_op_context(
+        op_config={
+        "repository_identifier": "simpl-open/platform-automation",
+        "consumer_email": "consumer@example.local",
+        "access_level": "read-only",
+        "token_name_prefix": "simpl-open",
+        "token_expiration_days": 30,
+        }
+    )
+
+    with pytest.raises(ValueError, match="GITEA_BASE_URL"):
+        generate_gitea_credentials(op_context)

tests/test_send_credentials_email.py

@@ -0,0 +1,74 @@
+"""Tests for email credential delivery."""
+
+import pytest
+
+from dagster import build_op_context
+
+from simpl_open_credential_delivery.ops.send_credentials_email import (
+    send_credentials_email,
+)
+from simpl_open_credential_delivery.services.email_service import (
+    EmailDeliveryService,
+)
+
+
+class FakeSMTP:
+    def __init__(self, host, port, timeout=None):
+        self.host = host
+        self.port = port
+        self.timeout = timeout
+        self.started_tls = False
+        self.logged_in = False
+        self.sent_messages = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def starttls(self):
+        self.started_tls = True
+
+    def login(self, username, password):
+        self.logged_in = True
+
+    def send_message(self, message):
+        self.sent_messages.append(message)
+
+
+def test_send_credentials_email_success(monkeypatch, op_context):
+    fake_smtp = FakeSMTP("smtp.example.local", 587)
+    monkeypatch.setenv("SMTP_HOST", "smtp.example.local")
+    monkeypatch.setenv("SMTP_PORT", "587")
+    monkeypatch.setenv("SMTP_FROM_ADDRESS", "no-reply@example.local")
+    monkeypatch.setenv("SMTP_USE_TLS", "true")
+    monkeypatch.setenv("EMAIL_SUBJECT_TEMPLATE", "Subject {repository_identifier}")
+    monkeypatch.setenv("EMAIL_BODY_TEMPLATE", "Hello {consumer_email} {access_token}")
+    monkeypatch.setattr("smtplib.SMTP", lambda host, port, timeout=None: fake_smtp)
+
+    result = send_credentials_email(
+        op_context,
+        {
+            "repository_identifier": "simpl-open/platform-automation",
+            "consumer_email": "consumer@example.local",
+            "access_level": "read-only",
+            "username": "application-consumer",
+            "access_token": "token-123",
+            "repository_access_url": "https://gitea.example.local/simpl-open/platform-automation",
+        },
+    )
+
+    assert result["email_delivery_status"] == "SENT"
+    assert result["email_message_id"] == ""
+    assert fake_smtp.started_tls is True
+    assert fake_smtp.sent_messages
+
+
+def test_email_delivery_service_requires_sender(monkeypatch):
+    monkeypatch.setenv("SMTP_HOST", "smtp.example.local")
+    monkeypatch.setenv("SMTP_PORT", "587")
+    monkeypatch.delenv("SMTP_FROM_ADDRESS", raising=False)
+
+    with pytest.raises(Exception, match="SMTP_FROM_ADDRESS"):
+        EmailDeliveryService.from_environment()

workspace.yaml

@@ -0,0 +1,4 @@
+load_from:
+  - python_module:
+      module_name: simpl_open_credential_delivery.repository
+      location_name: simpl_open_credential_delivery

yaml/values-dagster-simpl-open-credential-delivery-service.yaml

@@ -0,0 +1,4 @@
+dagster:
+  image:
+    repository: simpl-open-credential-delivery
+    tag: latest