awoooi/apps/api/tests/test_drift_detector_normalization.py

from __future__ import annotations

from copy import deepcopy

from src.models.drift import DriftLevel
from src.services.drift_detector import DriftDetector, GitStateReader


def test_git_state_reader_applies_kustomize_labels_and_images(tmp_path):
    prod_dir = tmp_path / "k8s" / "awoooi-prod"
    prod_dir.mkdir(parents=True)
    (prod_dir / "kustomization.yaml").write_text(
        """
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: awoooi-prod
commonLabels:
  environment: prod
  system: awoooi
resources:
  - deployment.yaml
images:
  - name: registry.local/library/api:IMAGE_TAG_PLACEHOLDER
    newName: registry.local/awoooi/api
    newTag: abc123
""",
        encoding="utf-8",
    )
    (prod_dir / "deployment.yaml").write_text(
        """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: awoooi-api
spec:
  selector:
    matchLabels:
      app: awoooi-api
  template:
    metadata:
      labels:
        app: awoooi-api
    spec:
      containers:
        - name: api
          image: registry.local/library/api:IMAGE_TAG_PLACEHOLDER
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 100
              podAffinityTerm:
                labelSelector:
                  matchLabels:
                    app: awoooi-api
                topologyKey: kubernetes.io/hostname
---
apiVersion: v1
kind: Service
metadata:
  name: awoooi-api-svc
spec:
  selector:
    app: awoooi-api
  ports:
    - port: 8000
      targetPort: 8000
""",
        encoding="utf-8",
    )

    state = GitStateReader(str(tmp_path / "k8s"))._read_sync("awoooi-prod")

    deployment = state["Deployment/awoooi-api"]
    template = deployment["spec"]["template"]
    assert deployment["metadata"]["namespace"] == "awoooi-prod"
    assert deployment["spec"]["selector"]["matchLabels"]["environment"] == "prod"
    assert template["metadata"]["labels"]["system"] == "awoooi"
    assert (
        template["spec"]["containers"][0]["image"]
        == "registry.local/awoooi/api:abc123"
    )
    affinity_selector = template["spec"]["affinity"]["podAntiAffinity"][
        "preferredDuringSchedulingIgnoredDuringExecution"
    ][0]["podAffinityTerm"]["labelSelector"]["matchLabels"]
    assert affinity_selector["environment"] == "prod"
    assert affinity_selector["system"] == "awoooi"

    service = state["Service/awoooi-api-svc"]
    assert service["metadata"]["namespace"] == "awoooi-prod"
    assert service["spec"]["selector"]["environment"] == "prod"
    assert service["spec"]["selector"]["system"] == "awoooi"


def test_kubernetes_api_defaults_do_not_create_drift():
    detector = DriftDetector()
    git_res = {
        "kind": "Deployment",
        "spec": {
            "selector": {"matchLabels": {"app": "awoooi-api"}},
            "template": {
                "metadata": {"labels": {"app": "awoooi-api"}},
                "spec": {
                    "serviceAccountName": "awoooi-executor",
                    "containers": [
                        {
                            "name": "api",
                            "image": "registry.local/awoooi/api:abc123",
                            "ports": [{"containerPort": 8000, "name": "http"}],
                            "readinessProbe": {
                                "httpGet": {"path": "/api/v1/health", "port": 8000},
                                "periodSeconds": 5,
                            },
                            "env": [
                                {
                                    "name": "POD_NAME",
                                    "valueFrom": {
                                        "fieldRef": {"fieldPath": "metadata.name"}
                                    },
                                }
                            ],
                        }
                    ],
                    "volumes": [
                        {"name": "runtime-secret", "secret": {"secretName": "s"}},
                        {"name": "runtime-config", "configMap": {"name": "c"}},
                    ],
                },
            },
        },
    }
    actual_res = deepcopy(git_res)
    pod_spec = actual_res["spec"]["template"]["spec"]
    pod_spec.update(
        {
            "serviceAccount": "awoooi-executor",
            "restartPolicy": "Always",
            "dnsPolicy": "ClusterFirst",
            "schedulerName": "default-scheduler",
            "terminationGracePeriodSeconds": 30,
            "securityContext": {},
        }
    )
    container = pod_spec["containers"][0]
    container["terminationMessagePath"] = "/dev/termination-log"
    container["terminationMessagePolicy"] = "File"
    container["ports"][0]["protocol"] = "TCP"
    container["readinessProbe"]["successThreshold"] = 1
    container["readinessProbe"]["httpGet"]["scheme"] = "HTTP"
    container["env"][0]["valueFrom"]["fieldRef"]["apiVersion"] = "v1"
    pod_spec["volumes"][0]["secret"]["defaultMode"] = 420
    pod_spec["volumes"][1]["configMap"]["defaultMode"] = 420
    actual_res["spec"]["template"]["metadata"]["annotations"] = {
        "kubectl.kubernetes.io/restartedAt": "2026-05-19T00:00:00+08:00"
    }

    assert (
        detector._diff_resources(
            git_res,
            actual_res,
            "Deployment",
            "awoooi-api",
            "awoooi-prod",
        )
        == []
    )


def test_service_runtime_defaults_do_not_create_drift():
    detector = DriftDetector()
    git_res = {
        "kind": "Service",
        "spec": {
            "type": "NodePort",
            "selector": {"app": "awoooi-api"},
            "ports": [{"port": 8000, "targetPort": 8000, "nodePort": 32334}],
        },
    }
    actual_res = deepcopy(git_res)
    actual_res["spec"].update(
        {
            "clusterIP": "10.43.156.119",
            "clusterIPs": ["10.43.156.119"],
            "ipFamilies": ["IPv4"],
            "ipFamilyPolicy": "SingleStack",
            "internalTrafficPolicy": "Cluster",
            "externalTrafficPolicy": "Cluster",
            "sessionAffinity": "None",
        }
    )
    actual_res["spec"]["ports"][0]["protocol"] = "TCP"

    assert (
        detector._diff_resources(
            git_res,
            actual_res,
            "Service",
            "awoooi-api-svc",
            "awoooi-prod",
        )
        == []
    )


def test_real_container_env_drift_remains_visible():
    detector = DriftDetector()
    git_res = {
        "kind": "Deployment",
        "spec": {
            "template": {
                "metadata": {"labels": {"app": "awoooi-api"}},
                "spec": {
                    "containers": [
                        {
                            "name": "api",
                            "image": "registry.local/awoooi/api:abc123",
                            "env": [{"name": "USE_AI_ROUTER", "value": "true"}],
                        }
                    ]
                },
            }
        },
    }
    actual_res = deepcopy(git_res)
    actual_res["spec"]["template"]["spec"]["containers"][0]["env"].append(
        {"name": "DUMMY_DEPLOY", "value": "1777914462"}
    )

    items = detector._diff_resources(
        git_res,
        actual_res,
        "Deployment",
        "awoooi-api",
        "awoooi-prod",
    )

    assert [item.field_path for item in items] == ["spec.template.spec.containers"]
    assert items[0].drift_level == DriftLevel.MEDIUM