acdl/tests/test_wiz_adapter.py

import json
import os
import sys
from pathlib import Path

import jsonschema
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from adapters.wiz.wiz_adapter import (
    SEVERITY_MAP, RESULT_MAP, _to_pcr, _emit_not_configured, adapt, is_configured,
)


FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"


class TestSeverityResultMaps:
    def test_severity_map_critical(self):
        assert SEVERITY_MAP["CRITICAL"] == "critical"
        assert SEVERITY_MAP["HIGH"] == "high"
        assert SEVERITY_MAP["MEDIUM"] == "medium"
        assert SEVERITY_MAP["LOW"] == "low"
        assert SEVERITY_MAP["INFO"] == "info"

    def test_result_map_open_is_fail(self):
        assert RESULT_MAP["OPEN"] == "fail"
        assert RESULT_MAP["RESOLVED"] == "pass"
        assert RESULT_MAP["IN_PROGRESS"] == "skipped"
        assert RESULT_MAP["DISMISSED"] == "skipped"


class TestToPcr:
    def test_translates_open_critical(self):
        issue = {
            "id": "wiz-1",
            "title": "a critical issue",
            "severity": "CRITICAL",
            "status": "OPEN",
            "entity": {"id": "arn:aws:s3:::b", "name": "b"},
        }
        pcr = _to_pcr(issue, "c-1")
        assert pcr["engine"] == "wiz"
        assert pcr["ruleId"] == "wiz-1"
        assert pcr["severity"] == "critical"
        assert pcr["result"] == "fail"
        assert pcr["contractId"] == "c-1"
        assert pcr["resourceRef"] == "arn:aws:s3:::b"

    def test_severity_case_insensitive(self):
        issue = {"id": "wiz-1", "severity": "high", "status": "open",
                 "entity": {"id": "r"}}
        pcr = _to_pcr(issue, "c-1")
        assert pcr["severity"] == "high"
        assert pcr["result"] == "fail"

    def test_unknown_severity_defaults_info(self):
        issue = {"id": "wiz-1", "severity": "BOGUS", "status": "OPEN",
                 "entity": {"id": "r"}}
        pcr = _to_pcr(issue, "c-1")
        assert pcr["severity"] == "info"

    def test_unknown_status_defaults_error(self):
        issue = {"id": "wiz-1", "severity": "INFO", "status": "BOGUS",
                 "entity": {"id": "r"}}
        pcr = _to_pcr(issue, "c-1")
        assert pcr["result"] == "error"

    def test_pcr_validates_against_schema(self, policy_check_result_schema):
        issue = {"id": "wiz-1", "title": "t", "severity": "CRITICAL",
                 "status": "OPEN", "entity": {"id": "r", "name": "n"}}
        pcr = _to_pcr(issue, "11111111-1111-1111-1111-111111111111")
        jsonschema.validate(pcr, policy_check_result_schema)


class TestEmitNotConfigured:
    def test_not_configured_pcr(self, policy_check_result_schema):
        pcr = _emit_not_configured("11111111-1111-1111-1111-111111111111")
        assert pcr["ruleId"] == "WIZ_NOT_CONFIGURED"
        assert pcr["result"] == "skipped"
        assert pcr["engine"] == "wiz"
        jsonschema.validate(pcr, policy_check_result_schema)


class TestAdapt:
    def test_translates_fixture(self, tmp_path, policy_check_result_schema):
        # adapt() reads a file path, copy fixture to a writable temp path
        src = FIXTURES / "wiz_issues.json"
        f = tmp_path / "wiz_issues.json"
        f.write_text(src.read_text())
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 3

        # issue 1: OPEN critical -> fail/critical; ruleId = control.name (v1.9 real client)
        assert results[0]["ruleId"] == "Public S3 bucket exposure"
        assert results[0]["severity"] == "critical"
        assert results[0]["result"] == "fail"

        # issue 2: RESOLVED high -> pass/high
        assert results[1]["ruleId"] == "Overly broad IAM role"
        assert results[1]["severity"] == "high"
        assert results[1]["result"] == "pass"

        # issue 3: IN_PROGRESS medium -> skipped/medium
        assert results[2]["ruleId"] == "SSH open to the world"
        assert results[2]["severity"] == "medium"
        assert results[2]["result"] == "skipped"

        for pcr in results:
            jsonschema.validate(pcr, policy_check_result_schema)

    def test_empty_issues_emits_not_configured(self, tmp_path):
        f = tmp_path / "wiz_empty.json"
        f.write_text(json.dumps({"issues": []}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 1
        assert results[0]["ruleId"] == "WIZ_NOT_CONFIGURED"
        assert results[0]["result"] == "skipped"

    def test_top_level_list_input(self, tmp_path):
        # data is a bare list (no "issues" wrapper)
        f = tmp_path / "wiz_list.json"
        f.write_text(json.dumps([
            {"id": "w-1", "severity": "LOW", "status": "RESOLVED",
             "entity": {"id": "r"}},
        ]))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 1
        assert results[0]["severity"] == "low"
        assert results[0]["result"] == "pass"


class TestIsConfigured:
    def test_not_configured_when_env_unset(self, monkeypatch):
        monkeypatch.delenv("WIZ_API_TOKEN", raising=False)
        assert is_configured() is False

    def test_configured_when_env_set(self, monkeypatch):
        monkeypatch.setenv("WIZ_API_TOKEN", "token-abc")
        monkeypatch.setenv("WIZ_API_URL", "https://api.wiz.io")
        assert is_configured() is True