import json

import pytest

import BlockDomain
from BlockDomain import (
    OBJECT_NAME_PREFIX,
    MAX_OBJECT_NAME_LENGTH,
    RESULT_FAILED,
    RESULT_SUCCESS,
    STATUS_DONE,
    STATUS_FAILED,
    PanOs,
    build_final_command_results,
    build_verbose_human_readable,
    derive_object_name,
    is_valid_fqdn,
    is_wildcard,
    normalize_tags,
    pan_os_commit,
    pan_os_commit_status,
    pan_os_push_status,
    pan_os_push_to_device,
    run_execute_command,
    validate_domains,
)


def ok_entry(entry_context=None, contents="ok", instance=None, brand="Panorama"):
    """Build a minimal successful execute_command entry.

    When ``instance`` is provided, includes a ``Metadata`` block matching what the platform
    actually returns (``Metadata.instance`` / ``Metadata.brand``) so tests can assert that the
    aggregate script correctly captures the serving-instance name from response entries.
    """
    entry: dict = {"Type": 1, "Contents": contents, "HumanReadable": "", "EntryContext": entry_context or {}}
    if instance is not None:
        entry["Metadata"] = {"instance": instance, "brand": brand}
    return entry


def err_entry(contents="error"):
    """Build a minimal error execute_command entry (Type 4 == entryTypes['error'])."""
    return {"Type": 4, "Contents": contents, "HumanReadable": "", "EntryContext": {}}


def test_run_execute_command_returns_entries(monkeypatch):
    """
    Given:
       - executeCommand returns a non-empty list of entries.
    When:
       - Calling run_execute_command.
    Then:
       - The entries are returned unchanged.
    """
    entries = [ok_entry()]
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: entries)

    assert run_execute_command("pan-os-get-address", {"name": "x"}) == entries


@pytest.mark.parametrize("empty_response", [[], None])
def test_run_execute_command_raises_on_empty_response(monkeypatch, empty_response):
    """
    Given:
       - executeCommand returns an empty response (empty list or None).
    When:
       - Calling run_execute_command.
    Then:
       - A DemistoException is raised instead of letting downstream res[0] indexing crash.
    """
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: empty_response)

    with pytest.raises(BlockDomain.DemistoException, match="returned an empty response"):
        run_execute_command("pan-os-get-address", {"name": "x"})


@pytest.mark.parametrize(
    "domain, expected",
    [
        ("*.evil.com", True),
        ("evil.*.com", True),
        ("evil.example.com", False),
        ("sub.domain.co.uk", False),
    ],
)
def test_is_wildcard(domain, expected):
    """
    Given:
       - A domain string that may or may not contain a wildcard character.
    When:
       - Calling is_wildcard to detect wildcard patterns.
    Then:
       - Returns True for domains containing '*', False otherwise.
    """
    assert is_wildcard(domain) is expected


@pytest.mark.parametrize(
    "domain, expected",
    [
        ("evil.example.com", True),
        ("sub.domain.co.uk", True),
        ("a.b", True),
        ("no-dot", False),
        ("*.evil.com", False),
        ("-leading.example.com", False),
        ("trailing-.example.com", False),
        ("space in.example.com", False),
        ("", False),
    ],
)
def test_is_valid_fqdn(domain, expected):
    """
    Given:
       - A domain string that may or may not be a syntactically valid FQDN.
    When:
       - Calling is_valid_fqdn to validate the domain.
    Then:
       - Returns True for well-formed FQDNs (has dot, valid labels, no wildcard/illegal chars),
         and False for anything else including empty strings and wildcards.
    """
    assert is_valid_fqdn(domain) is expected


def test_derive_object_name_is_deterministic():
    """
    Given:
       - The same FQDN passed to derive_object_name twice.
    When:
       - Comparing the two returned object names.
    Then:
       - Both invocations return the exact same string (deterministic mapping).
    """
    assert derive_object_name("evil.example.com") == derive_object_name("evil.example.com")


def test_derive_object_name_illegal_chars():
    """
    Given:
       - A domain containing an underscore, which is not a legal PAN-OS object-name character.
    When:
       - Calling derive_object_name.
    Then:
       - Underscores are normalized to hyphens so the resulting name is accepted by PAN-OS.
    """
    assert derive_object_name("bad_domain.example.com") == "Cortex-bad-domain.example.com"


def test_derive_object_name_overflow_truncates_and_hashes():
    """
    Given:
       - A domain long enough that the naive prefixed name would exceed the PAN-OS
         MAX_OBJECT_NAME_LENGTH limit.
    When:
       - Calling derive_object_name.
    Then:
       - The returned name fits within MAX_OBJECT_NAME_LENGTH, still starts with the
         "Cortex-" prefix, and remains deterministic across calls.
    """
    long_domain = ("a" * 80) + ".example.com"
    name = derive_object_name(long_domain)
    assert len(name) <= MAX_OBJECT_NAME_LENGTH
    assert name.startswith(OBJECT_NAME_PREFIX)


def test_derive_object_name_overflow_distinct_domains_are_unique():
    """
    Given:
       - Two DIFFERENT over-length domains.
    When:
       - Deriving each object name (both take the truncate+hash overflow path).
    Then:
       - The two names are different, proving the hash suffix disambiguates over-length names
         (the digest is a function of the full domain, not the truncated body).
    """
    name_a = derive_object_name(("a" * 80) + ".example.com")
    name_b = derive_object_name(("b" * 80) + ".example.com")
    assert name_a != name_b


def test_derive_object_name_overflow_same_prefix_different_tail_are_unique():
    """
    Given:
       - Two over-length domains that share an identical long leading segment (so their
         truncated bodies are the same) but differ only in a tail beyond the truncation point.
    When:
       - Deriving each object name.
    Then:
       - The names are STILL unique - the digest is computed over the full original domain, so
         differences past the truncation boundary are preserved in the hash suffix. This is the
         key anti-collision guarantee for the prefix-hash naming scheme.
    """
    shared_head = "a" * 80
    name_a = derive_object_name(f"{shared_head}.first-tail.example.com")
    name_b = derive_object_name(f"{shared_head}.second-tail.example.com")
    assert name_a != name_b
    # Both are truncated to the same length budget, so equal length; only the digest differs.
    assert len(name_a) == len(name_b)


def test_validate_domains_splits_valid_and_failed():
    """
    Given:
       - A mixed list of domains containing valid FQDNs, a wildcard, and an invalid FQDN.
    When:
       - Calling validate_domains to partition the input.
    Then:
       - Valid FQDNs are returned in the first list; the wildcard and invalid entries are
         returned as failed rows with STATUS_FAILED / RESULT_FAILED and appropriate messages.
    """
    valid, failed = validate_domains(["evil.example.com", "*.evil.com", "no-dot", "phish.attacker.net"])

    assert valid == ["evil.example.com", "phish.attacker.net"]
    assert len(failed) == 2

    wildcard_row = next(row for row in failed if row["Domain"] == "*.evil.com")
    assert wildcard_row["Status"] == STATUS_FAILED
    assert wildcard_row["Result"] == RESULT_FAILED
    assert "Wildcard" in wildcard_row["Message"]
    # No brand/rule applies to a validation-failed row - these must be real null, not "".
    assert wildcard_row["Brand"] is None
    assert wildcard_row["RuleName"] is None

    invalid_row = next(row for row in failed if row["Domain"] == "no-dot")
    assert invalid_row["Status"] == STATUS_FAILED
    assert invalid_row["Result"] == RESULT_FAILED
    assert "Invalid FQDN" in invalid_row["Message"]


def _pan_os(domains):
    return PanOs(
        {
            "domain_list": domains,
            "rule_name": "Cortex - Block Domain",
            "address_group": "Blocked Domains - Cortex",
            "tag": "cortex-blocked-domains",
            "log_forwarding_name": "",
            "auto_commit": True,
        }
    )


def _mock_execute(monkeypatch, responses_by_command):
    """Patch BlockDomain.demisto.executeCommand to answer per command name (order-independent).

    ``responses_by_command`` maps a command name to the entry list it should return. Probe commands
    (pan-os-get-address / -list-address-groups / -list-rules) supply the tenant state; write commands
    default to a bare success entry when not specified. ``pan-os-create-tag`` is always auto-answered.

    Args:
        monkeypatch: The pytest monkeypatch fixture.
        responses_by_command (dict): Mapping of command name -> entry list to return.
    Returns:
        The ``calls`` list of (command_name, args) tuples recorded during the run.
    """
    calls: list = []

    def _dispatch(command_name, args=None, *a, **k):
        calls.append((command_name, args or {}))
        if command_name == "pan-os-create-tag":
            return [ok_entry()]
        return responses_by_command.get(command_name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _dispatch)
    return calls


def test_process_domains_create_everything(monkeypatch):
    """
    Given:
       - A tenant where neither the address group nor the security rule exist yet, and the
         address object for the domain also does not exist.
    When:
       - Calling process_domains for a single domain.
    Then:
       - The address is created first, then the group is seeded with that object, then the
         rule is created against the now-existing group, then the rule is moved to the top.
         The resulting row is Done / Success / Created and carries the expected rule name.
    """
    _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-list-address-groups": [ok_entry({"Panorama.AddressGroups": []})],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
        },
    )
    pan = _pan_os(["evil.example.com"])
    rows = pan.process_domains()
    assert len(rows) == 1
    assert rows[0]["Status"] == STATUS_DONE
    assert rows[0]["Result"] == RESULT_SUCCESS
    assert rows[0]["RuleName"] == "Cortex - Block Domain"
    # Action is no longer surfaced; change-detection is tracked internally to gate the commit.
    assert pan._address_made_changes is True
    assert "Action" not in rows[0]


def test_ensure_tag_creates_tag_up_front(monkeypatch):
    """
    Given:
       - A PanOs flow configured with a tag.
    When:
       - Calling ensure_tag.
    Then:
       - pan-os-create-tag is invoked exactly once with the configured tag name, so the tag exists
         before any group/rule references it (independent of whether an address object is created).
    """
    calls: list = []

    def _capture(name, args):
        calls.append((name, args))
        return [ok_entry()]

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    _pan_os(["evil.example.com"]).ensure_tag()

    assert calls == [("pan-os-create-tag", {"name": "cortex-blocked-domains"})]


def test_ensure_tag_tolerates_already_exists_error(monkeypatch):
    """
    Given:
       - pan-os-create-tag returns an error (the tag already exists on the device).
    When:
       - Calling ensure_tag.
    Then:
       - ensure_tag does NOT raise - an already-existing tag is an expected, benign outcome.
    """

    def _err(name, args):
        return [err_entry("tag already exists")]

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _err)

    # Must not raise.
    _pan_os(["evil.example.com"]).ensure_tag()


def test_ensure_tag_noop_when_no_tag(monkeypatch):
    """
    Given:
       - A PanOs flow with an empty tag.
    When:
       - Calling ensure_tag.
    Then:
       - No pan-os-create-tag call is made (nothing to create).
    """
    calls: list = []

    def _capture(name, args):
        calls.append(name)
        return [ok_entry()]

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    pan = _pan_os(["evil.example.com"])
    pan.tag = ""
    pan.ensure_tag()

    assert calls == []


# ---- tag helpers (pure) ------------------------------------------------


@pytest.mark.parametrize(
    "raw, expected",
    [
        (None, []),
        ("", []),
        ("solo", ["solo"]),
        (["a", "b"], ["a", "b"]),
        (["a", "", None, "b"], ["a", "b"]),
    ],
)
def test_normalize_tags(raw, expected):
    """
    Given:
       - A raw PAN-OS 'Tags' value that may be None, an empty string, a single string, or a list
         possibly containing falsy entries.
    When:
       - Calling normalize_tags.
    Then:
       - It flattens the value into a clean list of tag-name strings, dropping falsy entries.
    """
    assert normalize_tags(raw) == expected


@pytest.mark.parametrize(
    "existing, expected",
    [
        ([], ["cortex-blocked-domains"]),
        (["other"], ["other", "cortex-blocked-domains"]),
        (["cortex-blocked-domains"], None),
        (["other", "cortex-blocked-domains"], None),
    ],
)
def test_merged_tags_if_missing(existing, expected):
    """
    Given:
       - An entity's current tag list and a PanOs flow configured with 'cortex-blocked-domains'.
    When:
       - Calling merged_tags_if_missing.
    Then:
       - Returns the de-duplicated union (existing + configured tag) when the tag is absent, and
         None (no edit needed) when the tag is already present. The existing tag is never duplicated.
    """
    assert _pan_os(["evil.example.com"]).merged_tags_if_missing(existing) == expected


def test_start_flow_skips_commit_when_all_actions_unchanged(monkeypatch):
    """
    Given:
       - A tenant already fully configured for the requested domain (group exists, rule exists,
         address object already a member) so every row is Unchanged.
    When:
       - Calling start_flow.
    Then:
       - pan_os_commit is never invoked; start_flow returns the Unchanged rows directly
         (skipping the multi-minute commit + push polling cycle on Panorama).
    """
    calls: list = []

    def _capture(name, args):
        calls.append((name, args))
        seq = {
            "pan-os-list-address-groups": [
                ok_entry(
                    {
                        "Panorama.AddressGroups": [
                            {
                                "Name": "Blocked Domains - Cortex",
                                "Type": "static",
                                "Addresses": ["Cortex-evil.example.com"],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            "pan-os-list-rules": [
                ok_entry(
                    {
                        "Panorama.SecurityRule": [
                            {
                                "Name": "Cortex - Block Domain",
                                "Destination": ["Blocked Domains - Cortex"],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            "pan-os-get-address": [
                ok_entry({"Panorama.Addresses": {"Name": "Cortex-evil.example.com", "Tags": ["cortex-blocked-domains"]}})
            ],
            "pan-os-move-rule": [ok_entry()],
        }
        return seq.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)
    monkeypatch.setattr(
        BlockDomain, "pan_os_commit", lambda *a, **k: pytest.fail("pan_os_commit must not be called when all rows are Unchanged")
    )
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda *a, **k: None)

    result = _pan_os(["evil.example.com"]).start_flow()

    assert isinstance(result, list)
    assert len(result) == 1
    assert "Action" not in result[0]
    assert "pan-os-commit" not in [c[0] for c in calls]


def test_start_flow_commits_when_at_least_one_row_modified(monkeypatch):
    """
    Given:
       - A tenant where the group and rule already exist but the address object for the
         requested domain is missing, so at least one row will end up Created.
    When:
       - Calling start_flow.
    Then:
       - pan_os_commit is invoked (candidate config was modified and must be pushed).
    """
    commit_called: list = []

    def _fake_commit(args, responses):
        commit_called.append(True)
        return BlockDomain.CommandResults(readable_output="fake commit ok")

    def _capture(name, args):
        seq = {
            "pan-os-list-address-groups": [
                ok_entry({"Panorama.AddressGroups": [{"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": []}]})
            ],
            "pan-os-list-rules": [
                ok_entry(
                    {"Panorama.SecurityRule": [{"Name": "Cortex - Block Domain", "Destination": ["Blocked Domains - Cortex"]}]}
                )
            ],
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-create-address": [ok_entry()],
            "pan-os-edit-address-group": [ok_entry()],
            "pan-os-move-rule": [ok_entry()],
        }
        return seq.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda *a, **k: None)
    monkeypatch.setattr(BlockDomain, "pan_os_commit", _fake_commit)
    monkeypatch.setattr(BlockDomain.demisto, "context", lambda: {"block_domain_rows": "[]"})

    _pan_os(["evil.example.com"]).start_flow()

    assert commit_called, "pan_os_commit must be called when at least one row is Created/Modified"


def test_process_domains_rule_created_when_object_unchanged_sets_rule_changed(monkeypatch):
    """
    Given:
       - The address object and group already exist and the object is already a member (so the
         per-domain Action is Unchanged), but the requested rule_name does NOT exist yet, so a
         new rule must be created.
    When:
       - Calling process_domains.
    Then:
       - The row Action is Unchanged (object/membership did not change), but the instance's
         _rule_changed flag is set True because pan-os-create-rule ran. This is what lets
         start_flow still commit a rule-only change.
    """
    obj = "Cortex-evil.example.com"

    def _capture(name, args):
        seq = {
            # Group exists, already contains the object, and already carries the tag -> Unchanged.
            "pan-os-list-address-groups": [
                ok_entry(
                    {
                        "Panorama.AddressGroups": [
                            {
                                "Name": "Blocked Domains - Cortex",
                                "Type": "static",
                                "Addresses": [obj],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            # No rule with the requested name exists -> ensure_rule will create it.
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
            # Object exists and already carries the tag -> no object tag edit.
            "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": ["cortex-blocked-domains"]}]})],
            "pan-os-create-rule": [ok_entry()],
            "pan-os-move-rule": [ok_entry()],
        }
        return seq.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    pan_os = _pan_os(["evil.example.com"])
    rows = pan_os.process_domains()

    # Object + membership did not change (so _made_changes stays False), but the rule was created,
    # so _rule_changed is True - that alone must still trigger a commit in start_flow.
    assert "Action" not in rows[0]
    assert pan_os._address_made_changes is False
    assert pan_os._rule_changed is True


def test_start_flow_commits_when_only_the_rule_changed(monkeypatch):
    """
    Given:
       - Every address object is already present and a member (all rows Unchanged), but a new
         rule had to be created this run (rule-only change).
    When:
       - Calling start_flow.
    Then:
       - pan_os_commit is still invoked, so the newly created rule is actually committed/pushed
         instead of silently sitting in the candidate config.
    """
    obj = "Cortex-evil.example.com"
    commit_called: list = []

    def _fake_commit(args, responses):
        commit_called.append(True)
        return BlockDomain.CommandResults(readable_output="fake commit ok")

    def _capture(name, args):
        seq = {
            "pan-os-list-address-groups": [
                ok_entry({"Panorama.AddressGroups": [{"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": [obj]}]})
            ],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
            "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj}]})],
            "pan-os-create-rule": [ok_entry()],
            "pan-os-move-rule": [ok_entry()],
        }
        return seq.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda *a, **k: None)
    monkeypatch.setattr(BlockDomain, "pan_os_commit", _fake_commit)
    monkeypatch.setattr(BlockDomain.demisto, "context", lambda: {"block_domain_rows": "[]"})

    _pan_os(["evil.example.com"]).start_flow()

    assert commit_called, "pan_os_commit must be called when only the rule changed (all objects Unchanged)"


def test_process_domains_missing_group_created_lazily_with_first_object(monkeypatch):
    """
    Given:
       - A tenant with neither group nor rule pre-configured, and no address object for the
         requested domain. PAN-OS refuses to create a static group with no members, and
         refuses to create a rule whose destination does not resolve to an existing object.
    When:
       - Calling process_domains for a single domain.
    Then:
       - The address is created first; then the group is created lazily seeded with that
         first object (never as an empty static group); then the rule is created against the
         now-existing group. No edit-address-group call is emitted for the seed domain.
    """
    calls: list = []

    def _capture(name, args):
        calls.append((name, args))
        seq = {
            "pan-os-list-address-groups": [ok_entry({"Panorama.AddressGroups": []})],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
            "pan-os-create-rule": [ok_entry()],
            "pan-os-move-rule": [ok_entry()],
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-create-address": [ok_entry()],
            "pan-os-create-address-group": [ok_entry()],
        }
        return seq.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    _pan_os(["evil.example.com"]).process_domains()

    names = [c[0] for c in calls]
    assert names.index("pan-os-create-address") < names.index("pan-os-create-address-group")
    assert names.index("pan-os-create-address-group") < names.index("pan-os-create-rule")
    group_create_args = next(args for name, args in calls if name == "pan-os-create-address-group")
    assert group_create_args["type"] == "static"
    # The group is created once with the full member list (all objects created first).
    assert group_create_args["addresses"] == ["Cortex-evil.example.com"]
    rule_create_args = next(args for name, args in calls if name == "pan-os-create-rule")
    assert rule_create_args["destination"] == "Blocked Domains - Cortex"
    assert "pan-os-edit-address-group" not in names


def test_process_domains_multi_domain_creates_all_objects_before_group(monkeypatch):
    """
    Given:
       - A tenant with no group, no rule, and neither address object existing, for TWO domains.
    When:
       - Calling process_domains for both domains.
    Then:
       - Both address objects are created BEFORE the single group-create, and the group is created
         once with the full member list (this is the ordering that avoids the reference error).
    """
    calls = _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-list-address-groups": [ok_entry({"Panorama.AddressGroups": []})],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
        },
    )
    _pan_os(["evil.example.com", "phish.attacker.net"]).process_domains()

    names = [name for name, _ in calls]
    create_addresses = [i for i, name in enumerate(names) if name == "pan-os-create-address"]
    group_create_index = names.index("pan-os-create-address-group")
    # Both objects are created before the single group-create.
    assert len(create_addresses) == 2
    assert max(create_addresses) < group_create_index
    assert names.count("pan-os-create-address-group") == 1
    group_args = next(args for name, args in calls if name == "pan-os-create-address-group")
    assert group_args["addresses"] == ["Cortex-evil.example.com", "Cortex-phish.attacker.net"]


def test_process_domains_all_unchanged(monkeypatch):
    """
    Given:
       - A tenant where the group, rule, and address object all already exist and the
         address object is already a member of the group.
    When:
       - Calling process_domains.
    Then:
       - The resulting row is Done / Unchanged (no create/modify happened).
    """
    obj = "Cortex-evil.example.com"
    calls = _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": ["cortex-blocked-domains"]}]})],
            "pan-os-list-address-groups": [
                ok_entry(
                    {
                        "Panorama.AddressGroups": [
                            {
                                "Name": "Blocked Domains - Cortex",
                                "Type": "static",
                                "Addresses": [obj],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            "pan-os-list-rules": [
                ok_entry(
                    {
                        "Panorama.SecurityRule": [
                            {
                                "Name": "Cortex - Block Domain",
                                "Destination": ["Blocked Domains - Cortex"],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
        },
    )
    pan = _pan_os(["evil.example.com"])
    rows = pan.process_domains()
    assert rows[0]["Status"] == STATUS_DONE
    # Everything already in place (incl. the tag) -> no change -> commit will be skipped.
    assert pan._address_made_changes is False
    assert pan._rule_changed is False
    assert "Action" not in rows[0]
    # A pre-existing, unchanged rule must NOT be moved (no needless candidate-config churn).
    assert not any(name == "pan-os-move-rule" for name, _ in calls)


def test_process_domains_modified_when_added_to_existing_group(monkeypatch):
    """
    Given:
       - A tenant where the group and rule already exist but the group has no members yet
         and the address object for the requested domain does not exist.
    When:
       - Calling process_domains.
    Then:
       - The address is created and added to the existing group; the resulting row is
         Done / Created (most-significant action of the create-address step).
    """
    _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-list-address-groups": [
                ok_entry(
                    {
                        "Panorama.AddressGroups": [
                            {
                                "Name": "Blocked Domains - Cortex",
                                "Type": "static",
                                "Addresses": [],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            "pan-os-list-rules": [
                ok_entry(
                    {
                        "Panorama.SecurityRule": [
                            {
                                "Name": "Cortex - Block Domain",
                                "Destination": ["Blocked Domains - Cortex"],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
        },
    )
    pan = _pan_os(["evil.example.com"])
    rows = pan.process_domains()
    assert rows[0]["Status"] == STATUS_DONE
    # A new object was created and added to the group -> a change occurred -> commit will run.
    assert pan._address_made_changes is True
    assert "Action" not in rows[0]


def test_process_domains_existing_rule_missing_group_is_edited(monkeypatch):
    """
    Given:
       - A tenant where the rule exists but its destination does not yet reference our
         address group; the group and the address object already exist.
    When:
       - Calling process_domains.
    Then:
       - The rule is edited to include the group in its destination; the resulting row is
         Done / Unchanged because the address object and its group membership were unchanged
         (rule-level fix is a group-scope, not per-domain, mutation).
    """
    obj = "Cortex-evil.example.com"
    _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": ["cortex-blocked-domains"]}]})],
            "pan-os-list-address-groups": [
                ok_entry(
                    {
                        "Panorama.AddressGroups": [
                            {
                                "Name": "Blocked Domains - Cortex",
                                "Type": "static",
                                "Addresses": [obj],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
            "pan-os-list-rules": [
                ok_entry(
                    {
                        "Panorama.SecurityRule": [
                            {
                                "Name": "Cortex - Block Domain",
                                "Destination": ["something-else"],
                                "Tags": ["cortex-blocked-domains"],
                            }
                        ]
                    }
                )
            ],
        },
    )
    pan = _pan_os(["evil.example.com"])
    rows = pan.process_domains()
    assert rows[0]["Status"] == STATUS_DONE
    # Object + membership unchanged, but the rule was edited to add the group -> _rule_changed.
    assert pan._address_made_changes is False
    assert pan._rule_changed is True
    assert "Action" not in rows[0]


def _tag_capture(monkeypatch):
    """Patch executeCommand with a dict-based capture that tolerates call ordering.

    Returns the ``calls`` list of (command_name, args) tuples. Read/probe commands return
    fixtures that model an already-existing object/group/rule WITHOUT the configured tag, so the
    merge/edit paths are exercised. Every other command returns a bare success entry.
    """
    obj = "Cortex-evil.example.com"
    calls: list = []
    fixtures = {
        "pan-os-list-address-groups": [
            ok_entry(
                {
                    "Panorama.AddressGroups": [
                        {"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": [obj], "Tags": ["other"]}
                    ]
                }
            )
        ],
        "pan-os-list-rules": [
            ok_entry(
                {
                    "Panorama.SecurityRule": [
                        {"Name": "Cortex - Block Domain", "Destination": ["Blocked Domains - Cortex"], "Tags": ["other"]}
                    ]
                }
            )
        ],
        "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": ["other"]}]})],
    }

    def _capture(name, args):
        calls.append((name, args))
        return fixtures.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)
    return calls


def test_process_domains_merges_tag_on_existing_object_group_and_rule(monkeypatch):
    """
    Given:
       - A tenant where the address object, address group, and rule all already exist, each
         carrying a different pre-existing tag ('other') but NOT the configured tag.
    When:
       - Calling process_domains with tag='cortex-blocked-domains'.
    Then:
       - The configured tag is merged (never replacing) onto all three entities:
         * pan-os-edit-address with element_to_change=tag and the merged list ['other', tag]
         * pan-os-edit-address-group with tags=['other', tag]
         * pan-os-edit-rule with element_to_change=tag, behaviour=add, element_value=tag
    """
    calls = _tag_capture(monkeypatch)
    pan = _pan_os(["evil.example.com"])
    rows = pan.process_domains()

    assert rows[0]["Status"] == STATUS_DONE

    edit_address = [args for name, args in calls if name == "pan-os-edit-address"]
    assert edit_address == [
        {
            "name": "Cortex-evil.example.com",
            "element_to_change": "tag",
            "element_value": ["other", "cortex-blocked-domains"],
        }
    ]

    # The group edit carries element_to_add (required by the static-group guard) together with the
    # merged tags in one call. The object is already a member, so the full list is re-added (no-op).
    edit_group = [args for name, args in calls if name == "pan-os-edit-address-group" and "tags" in args]
    assert edit_group == [
        {
            "name": "Blocked Domains - Cortex",
            "type": "static",
            "element_to_add": ["Cortex-evil.example.com"],
            "tags": ["other", "cortex-blocked-domains"],
        }
    ]

    edit_rule_tag = [args for name, args in calls if name == "pan-os-edit-rule" and args.get("element_to_change") == "tag"]
    assert edit_rule_tag == [
        {
            "rulename": "Cortex - Block Domain",
            "element_to_change": "tag",
            "element_value": "cortex-blocked-domains",
            "behaviour": "add",
            "pre_post": "pre-rulebase",
        }
    ]


def test_process_domains_does_not_re_tag_when_tag_already_present(monkeypatch):
    """
    Given:
       - The demo scenario: an existing FQDN, an existing group, and an existing rule that ALL
         already carry the configured tag; the user re-runs the command.
    When:
       - Calling process_domains with the same tag.
    Then:
       - No tag edit is issued on the object, group, or rule (the tag is never duplicated), and
         nothing changes so the commit is skipped.
    """
    obj = "Cortex-evil.example.com"
    tag = "cortex-blocked-domains"
    calls: list = []
    fixtures = {
        "pan-os-list-address-groups": [
            ok_entry(
                {
                    "Panorama.AddressGroups": [
                        {"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": [obj], "Tags": [tag]}
                    ]
                }
            )
        ],
        "pan-os-list-rules": [
            ok_entry(
                {
                    "Panorama.SecurityRule": [
                        {"Name": "Cortex - Block Domain", "Destination": ["Blocked Domains - Cortex"], "Tags": [tag]}
                    ]
                }
            )
        ],
        "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": [tag]}]})],
    }

    def _capture(name, args):
        calls.append((name, args))
        return fixtures.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    pan = _pan_os(["evil.example.com"])
    pan.process_domains()

    assert [args for name, args in calls if name == "pan-os-edit-address"] == []
    assert [args for name, args in calls if name == "pan-os-edit-address-group" and "tags" in args] == []
    assert [args for name, args in calls if name == "pan-os-edit-rule" and args.get("element_to_change") == "tag"] == []
    assert pan._address_made_changes is False
    assert pan._rule_changed is False


def test_process_domains_new_rule_name_does_not_re_tag_existing_object_and_group(monkeypatch):
    """
    Given:
       - An existing FQDN and existing group that ALREADY carry the configured tag, but a brand-new
         rule name that does not exist yet (the "just add a rule" case).
    When:
       - Calling process_domains.
    Then:
       - The object and group are NOT re-tagged (no duplication); only the new rule is created
         (which carries the tag at creation time via its own create args).
    """
    obj = "Cortex-evil.example.com"
    tag = "cortex-blocked-domains"
    calls: list = []
    fixtures = {
        "pan-os-list-address-groups": [
            ok_entry(
                {
                    "Panorama.AddressGroups": [
                        {"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": [obj], "Tags": [tag]}
                    ]
                }
            )
        ],
        # Rule does not exist yet -> it will be created (with the tag in its create args).
        "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
        "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": [tag]}]})],
    }

    def _capture(name, args):
        calls.append((name, args))
        return fixtures.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)

    pan = _pan_os(["evil.example.com"])
    pan.process_domains()

    # Object and group already carry the tag -> no re-tag edits.
    assert [args for name, args in calls if name == "pan-os-edit-address"] == []
    assert [args for name, args in calls if name == "pan-os-edit-address-group" and "tags" in args] == []
    # The new rule is created and carries the tag at creation.
    create_rule = [args for name, args in calls if name == "pan-os-create-rule"]
    assert len(create_rule) == 1
    assert create_rule[0]["tags"] == tag
    assert pan._rule_changed is True


def _pan_os_with_log_forwarding(domains, profile):
    """Build a PanOs flow configured with a log-forwarding profile name (rule-only)."""
    pan = _pan_os(domains)
    pan.log_forwarding_name = profile
    return pan


def _lf_capture(monkeypatch, existing_profile):
    """Patch executeCommand modelling an existing object/group/rule (everything already in place).

    The existing rule carries ``existing_profile`` as its LogForwardingProfile. Returns the
    ``calls`` list of (command_name, args) tuples. Tags are pre-present so no tag edits fire.
    """
    obj = "Cortex-evil.example.com"
    tag = "cortex-blocked-domains"
    calls: list = []
    fixtures = {
        "pan-os-list-address-groups": [
            ok_entry(
                {
                    "Panorama.AddressGroups": [
                        {"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": [obj], "Tags": [tag]}
                    ]
                }
            )
        ],
        "pan-os-list-rules": [
            ok_entry(
                {
                    "Panorama.SecurityRule": [
                        {
                            "Name": "Cortex - Block Domain",
                            "Destination": ["Blocked Domains - Cortex"],
                            "Tags": [tag],
                            "LogForwardingProfile": existing_profile,
                        }
                    ]
                }
            )
        ],
        "pan-os-get-address": [ok_entry({"Panorama.Addresses": [{"Name": obj, "Tags": [tag]}]})],
    }

    def _capture(name, args):
        calls.append((name, args))
        return fixtures.get(name, [ok_entry()])

    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", _capture)
    return calls


def test_process_domains_sets_log_forwarding_on_existing_rule_when_absent(monkeypatch):
    """
    Given:
       - An existing rule with NO log-forwarding profile, and a flow configured with one.
    When:
       - Calling process_domains.
    Then:
       - pan-os-edit-rule is issued with element_to_change=log-forwarding and the configured
         profile, and _rule_changed is set.
    """
    calls = _lf_capture(monkeypatch, existing_profile="")
    pan = _pan_os_with_log_forwarding(["evil.example.com"], "cortex-lfp")
    pan.process_domains()

    edit_lf = [args for name, args in calls if name == "pan-os-edit-rule" and args.get("element_to_change") == "log-forwarding"]
    assert edit_lf == [
        {
            "rulename": "Cortex - Block Domain",
            "element_to_change": "log-forwarding",
            "element_value": "cortex-lfp",
            "pre_post": "pre-rulebase",
        }
    ]
    assert pan._rule_changed is True


def test_process_domains_sets_log_forwarding_on_existing_rule_when_different(monkeypatch):
    """
    Given:
       - An existing rule whose log-forwarding profile differs from the configured one.
    When:
       - Calling process_domains.
    Then:
       - pan-os-edit-rule replaces it with the configured profile (single-value, no dedup concept).
    """
    calls = _lf_capture(monkeypatch, existing_profile="old-lfp")
    pan = _pan_os_with_log_forwarding(["evil.example.com"], "cortex-lfp")
    pan.process_domains()

    edit_lf = [args for name, args in calls if name == "pan-os-edit-rule" and args.get("element_to_change") == "log-forwarding"]
    assert len(edit_lf) == 1
    assert edit_lf[0]["element_value"] == "cortex-lfp"
    assert pan._rule_changed is True


def test_process_domains_does_not_set_log_forwarding_when_already_matches(monkeypatch):
    """
    Given:
       - An existing rule that ALREADY carries the configured log-forwarding profile.
    When:
       - Calling process_domains.
    Then:
       - No log-forwarding edit is issued (no unnecessary edit), and since nothing else changed
         the rule is left untouched.
    """
    calls = _lf_capture(monkeypatch, existing_profile="cortex-lfp")
    pan = _pan_os_with_log_forwarding(["evil.example.com"], "cortex-lfp")
    pan.process_domains()

    edit_lf = [args for name, args in calls if name == "pan-os-edit-rule" and args.get("element_to_change") == "log-forwarding"]
    assert edit_lf == []
    assert pan._rule_changed is False


def test_process_domains_dynamic_group_is_failed(monkeypatch):
    """
    Given:
       - A tenant where the target group already exists but as a *dynamic* address group
         (which cannot accept manually added static members).
    When:
       - Calling process_domains.
    Then:
       - The domain could not be blocked, so the row is marked Failed / Failed and the message
         explains the group is dynamic.
    """
    _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-list-address-groups": [
                ok_entry({"Panorama.AddressGroups": [{"Name": "Blocked Domains - Cortex", "Type": "dynamic", "Match": "x"}]})
            ],
        },
    )
    # Capture the abort log so it does not leak to stdout (conftest fails on any stdout).
    errors: list = []
    monkeypatch.setattr(BlockDomain.demisto, "error", lambda msg: errors.append(msg))

    rows = _pan_os(["evil.example.com"]).process_domains()
    assert rows[0]["Status"] == STATUS_FAILED
    assert rows[0]["Result"] == RESULT_FAILED
    assert "dynamic" in rows[0]["Message"]
    # The aborted domains are logged for troubleshooting.
    assert any("evil.example.com" in msg for msg in errors)


def test_process_domains_moves_rule_to_top_when_rule_created(monkeypatch):
    """
    Given:
       - A tenant where neither the rule nor the address object exist yet, so the rule is created.
    When:
       - Calling process_domains.
    Then:
       - pan-os-move-rule is issued to enforce top placement for the newly created rule.
    """
    calls = _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-list-address-groups": [
                ok_entry({"Panorama.AddressGroups": [{"Name": "Blocked Domains - Cortex", "Type": "static", "Addresses": []}]})
            ],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
        },
    )
    pan = _pan_os(["evil.example.com"])
    pan.process_domains()
    assert pan._rule_changed is True
    assert any(name == "pan-os-move-rule" for name, _ in calls)


def test_process_domains_failure_marks_row_failed(monkeypatch):
    """
    Given:
       - A tenant where the group and rule do not exist yet, and the pan-os-create-address
         command fails (e.g. permission denied) before the group and rule are touched.
    When:
       - Calling process_domains.
    Then:
       - The resulting row is Failed / Failed and the error message from PAN-OS is surfaced,
         and the full traceback is logged via demisto.error.
    """
    _mock_execute(
        monkeypatch,
        {
            "pan-os-get-address": [err_entry("not found")],
            "pan-os-create-address": [err_entry("permission denied")],
            "pan-os-list-address-groups": [ok_entry({"Panorama.AddressGroups": []})],
            "pan-os-list-rules": [ok_entry({"Panorama.SecurityRule": []})],
        },
    )
    # Capture the traceback log so it does not leak to stdout (conftest fails on any stdout).
    errors: list = []
    monkeypatch.setattr(BlockDomain.demisto, "error", lambda msg: errors.append(msg))

    rows = _pan_os(["evil.example.com"]).process_domains()
    assert rows[0]["Status"] == STATUS_FAILED
    assert rows[0]["Result"] == RESULT_FAILED
    assert "permission denied" in rows[0]["Message"]
    assert any("process_domains failed" in msg for msg in errors)


def test_build_verbose_human_readable_joins_with_blank_lines():
    """
    Given:
       - A list of response entries where some carry a HumanReadable string and some do not.
    When:
       - Calling build_verbose_human_readable.
    Then:
       - Entries without HumanReadable are skipped; the rest are joined with a blank line
         separator, prefixed by a leading blank line so the block detaches from the summary
         table above it.
    """
    responses = [
        [ok_entry(contents="c1")],
        [{"Type": 1, "Contents": "c2", "HumanReadable": "HR-two", "EntryContext": {}}],
        [{"Type": 1, "Contents": "c3", "HumanReadable": "HR-three", "EntryContext": {}}],
    ]
    verbose_hr = build_verbose_human_readable(responses)
    assert verbose_hr == "\n\nHR-two\n\nHR-three"


def test_build_verbose_human_readable_empty_when_no_hr():
    """
    Given:
       - A response list where no entry has a HumanReadable string, or an empty list.
    When:
       - Calling build_verbose_human_readable.
    Then:
       - Returns an empty string (nothing to append to the summary table).
    """
    assert build_verbose_human_readable([[ok_entry(contents="c1")]]) == ""
    assert build_verbose_human_readable([]) == ""


def test_build_final_command_results_non_verbose_is_table_only():
    """
    Given:
       - A rows list and a set of responses that include HumanReadable content, with
         verbose=False.
    When:
       - Calling build_final_command_results.
    Then:
       - The returned CommandResults uses the BlockDomain context prefix, exposes the
         rows unchanged as outputs, renders the summary table containing the domain, and does
         NOT append any of the per-command verbose HR blocks.
    """
    rows = [
        {
            "Domain": "a.com",
            "Brand": "Panorama",
            "Status": STATUS_DONE,
            "Result": RESULT_SUCCESS,
            "RuleName": "Cortex - Block Domain",
            "Message": "ok",
        }
    ]
    responses = [[{"Type": 1, "Contents": "c", "HumanReadable": "HR", "EntryContext": {}}]]

    result = build_final_command_results(rows, verbose=False, responses=responses)
    assert result.outputs_prefix == "BlockDomain"
    assert result.outputs == rows
    assert "a.com" in result.readable_output
    assert "HR" not in result.readable_output


def test_build_final_command_results_verbose_appends_command_hr():
    """
    Given:
       - A rows list and responses with a HumanReadable block, with verbose=True.
    When:
       - Calling build_final_command_results.
    Then:
       - The returned CommandResults exposes the rows as outputs, renders the summary table
         containing the domain, and appends the per-command HR block at the end of the
         readable_output (so users can see exactly what each downstream call produced).
    """
    rows = [
        {
            "Domain": "a.com",
            "Brand": "Panorama",
            "Status": STATUS_DONE,
            "Result": RESULT_SUCCESS,
            "RuleName": "Cortex - Block Domain",
            "Message": "ok",
        }
    ]
    responses = [[{"Type": 1, "Contents": "c", "HumanReadable": "HR-one", "EntryContext": {}}]]

    result = build_final_command_results(rows, verbose=True, responses=responses)
    assert result.outputs == rows
    assert "a.com" in result.readable_output
    assert result.readable_output.endswith("HR-one")


def test_build_final_command_results_all_failed_is_error_entry():
    """
    Given:
       - A rows list where every row has Result == Failed.
    When:
       - Calling build_final_command_results.
    Then:
       - The returned CommandResults is marked as an ERROR entry, still exposes the rows as
         outputs, and its readable_output carries the "All runs failed" header so the war room
         and any downstream playbook can branch on the failure.
    """
    rows = [
        {
            "Domain": "a.com",
            "Brand": "Panorama",
            "Status": STATUS_FAILED,
            "Result": RESULT_FAILED,
            "RuleName": "",
            "Message": "boom",
        },
        {
            "Domain": "b.com",
            "Brand": "Panorama",
            "Status": STATUS_FAILED,
            "Result": RESULT_FAILED,
            "RuleName": "",
            "Message": "boom",
        },
    ]

    result = build_final_command_results(rows, verbose=False, responses=[])
    assert result.entry_type == BlockDomain.EntryType.ERROR
    assert result.outputs == rows
    assert "All runs failed" in result.readable_output


def test_build_final_command_results_partial_failure_is_not_error_entry():
    """
    Given:
       - A rows list with at least one successful row alongside a failed row.
    When:
       - Calling build_final_command_results.
    Then:
       - The returned CommandResults is NOT an ERROR entry (entry_type is unset / not ERROR),
         so a partial success is still reported as a normal result.
    """
    rows = [
        {
            "Domain": "a.com",
            "Brand": "Panorama",
            "Status": STATUS_DONE,
            "Result": RESULT_SUCCESS,
            "RuleName": "Cortex - Block Domain",
            "Message": "ok",
        },
        {
            "Domain": "b.com",
            "Brand": "Panorama",
            "Status": STATUS_FAILED,
            "Result": RESULT_FAILED,
            "RuleName": "",
            "Message": "boom",
        },
    ]

    result = build_final_command_results(rows, verbose=False, responses=[])
    assert result.entry_type != BlockDomain.EntryType.ERROR
    assert "All runs failed" not in result.readable_output


def test_build_final_command_results_empty_rows_is_not_error_entry():
    """
    Given:
       - An empty rows list (e.g. no domains reached a vendor).
    When:
       - Calling build_final_command_results.
    Then:
       - The run is NOT reported as an error (an empty run is not an all-failed run).
    """
    result = build_final_command_results([], verbose=False, responses=[])
    assert result.entry_type != BlockDomain.EntryType.ERROR
    assert "All runs failed" not in result.readable_output


def test_pan_os_commit_status_reports_pending_while_job_active(monkeypatch):
    """
    Given:
       - pan-os-commit-status returns a job with result 'OK' but status 'ACT' (still running).
    When:
       - Calling pan_os_commit_status.
    Then:
       - The war-room output must NOT show 'Success' while the job is still active; it reports
         'Pending' (job.result 'OK' and job.status 'ACT' are independent PAN-OS fields, and success
         is only reported once the job status is 'FIN').
    """
    active_entry = {
        "Type": 1,
        "Contents": {"response": {"result": {"job": {"result": "OK", "status": "ACT"}}}},
        "HumanReadable": "",
        "EntryContext": {},
    }
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: [active_entry])

    result = pan_os_commit_status({"commit_job_id": "123"}, [])

    assert result.outputs == {"JobID": "123", "Status": "Pending"}  # type: ignore[attr-defined]


def test_pan_os_commit_status_reports_success_only_when_finished(monkeypatch):
    """
    Given:
       - pan-os-commit-status returns a finished job (status 'FIN') with result 'OK'.
    When:
       - Calling pan_os_commit_status.
    Then:
       - The output reports 'Success' and polling stops.
    """
    fin_entry = {
        "Type": 1,
        "Contents": {"response": {"result": {"job": {"result": "OK", "status": "FIN"}}}},
        "HumanReadable": "",
        "EntryContext": {},
    }
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: [fin_entry])

    result = pan_os_commit_status({"commit_job_id": "123"}, [])

    assert result.outputs == {"JobID": "123", "Status": "Success"}  # type: ignore[attr-defined]
    assert not BlockDomain.is_polling_in_progress(result)


def test_pan_os_commit_reports_failure_on_error(monkeypatch):
    """
    Given:
       - pan-os-commit returns an error entry.
    When:
       - Calling pan_os_commit.
    Then:
       - The failure is surfaced (Status 'Failure') and polling stops, instead of being silently
         treated as 'no changes to commit'.
    """
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: [err_entry("commit failed")])
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda *a, **k: None)

    result = pan_os_commit({}, [])

    assert result.outputs == {"Status": "Failure"}  # type: ignore[attr-defined]
    assert not BlockDomain.is_polling_in_progress(result)


def test_pan_os_push_to_device_reports_failure_on_error(monkeypatch):
    """
    Given:
       - pan-os-push-to-device-group returns an error entry.
    When:
       - Calling pan_os_push_to_device.
    Then:
       - The failure is surfaced (Status 'Failure') and polling stops, instead of being silently
         treated as 'no changes to push'.
    """
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: [err_entry("push failed")])
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda *a, **k: None)

    result = pan_os_push_to_device({}, [])

    assert result.outputs == {"Status": "Failure"}  # type: ignore[attr-defined]
    assert not BlockDomain.is_polling_in_progress(result)


def test_pan_os_push_status_error_contents_is_terminal_failure(monkeypatch):
    """
    Given:
       - pan-os-push-status returns an error entry whose Contents is a plain string
         (PAN-OS surfaces errors as a bare string instead of the nested status dict).
    When:
       - Calling pan_os_push_status.
    Then:
       - The function does NOT crash with 'str object has no attribute get'; it stops polling
         (no ScheduledCommand attached) and reports a Failure status for the job.
    """
    monkeypatch.setattr(
        BlockDomain.demisto,
        "executeCommand",
        lambda *a, **k: [err_entry("Failed to execute pan-os-push-status. Error: job not found")],
    )

    # The @polling_function decorator unwraps the PollResult and returns its CommandResults response
    # at runtime, though the declared return type is still PollResult (hence the type: ignore below).
    result = pan_os_push_status({"push_job_id": "123"}, [])

    assert not BlockDomain.is_polling_in_progress(result)
    assert result.outputs == {"JobID": "123", "Status": "Failure"}  # type: ignore[attr-defined]


def test_pan_os_push_status_fin_stops_polling(monkeypatch):
    """
    Given:
       - pan-os-push-status returns a well-formed nested dict whose job status is 'FIN'.
    When:
       - Calling pan_os_push_status.
    Then:
       - Polling stops (no ScheduledCommand attached) and the reported job status is 'FIN'.
    """
    fin_entry = {
        "Type": 1,
        "Contents": {"response": {"result": {"job": {"status": "FIN"}}}},
        "HumanReadable": "",
        "EntryContext": {},
    }
    monkeypatch.setattr(BlockDomain.demisto, "executeCommand", lambda *a, **k: [fin_entry])

    result = pan_os_push_status({"push_job_id": "456"}, [])

    assert not BlockDomain.is_polling_in_progress(result)
    assert result.outputs == {"Status": "FIN", "JobID": "456"}  # type: ignore[attr-defined]


def _install_fake_context(monkeypatch):
    """Back demisto.context()/setContext() with an in-memory dict, mirroring platform behavior.

    Returns the backing store so tests can inspect exactly what was serialized to context.
    """
    store: dict = {}
    monkeypatch.setattr(BlockDomain.demisto, "context", lambda: dict(store))
    monkeypatch.setattr(BlockDomain.demisto, "setContext", lambda key, value: store.__setitem__(key, value))
    return store


def test_save_and_restore_responses_json_round_trip(monkeypatch):
    """
    Given:
       - A PanOs instance whose accumulated responses contain the realistic PAN-OS entry shape
         (nested Contents dict, Metadata, None HumanReadable) written to context as JSON.
    When:
       - Calling save_responses on one polling cycle and restore_responses on the next.
    Then:
       - The context value is valid JSON (not a Python repr), and the responses survive the
         json.dumps -> json.loads round-trip byte-for-byte equal to the reduced form.
    """
    store = _install_fake_context(monkeypatch)

    pan_os = _pan_os(["evil.example.com"])
    pan_os.responses = [
        [
            {
                "Type": 1,
                "Contents": {"response": {"result": {"job": {"status": "FIN", "id": "42"}}}},
                "HumanReadable": None,
                "Metadata": {"instance": "Panorama_QA", "brand": "Panorama"},
                "EntryContext": {"dropped": "not serialized"},
            }
        ]
    ]
    expected_reduced = pan_os.reduce_responses()

    pan_os.save_responses()

    # Stored value must be real JSON that json.loads can parse (would fail on a Python repr).
    stored = store["panorama_responses"]
    assert json.loads(stored) == expected_reduced

    # A fresh instance restoring from the same context recovers the reduced responses exactly.
    fresh = _pan_os(["evil.example.com"])
    fresh.restore_responses()
    assert fresh.responses == expected_reduced


def test_finish_reads_rows_and_responses_as_json_then_clears_context(monkeypatch):
    """
    Given:
       - Context holds block_domain_rows and panorama_responses that were written as JSON by a
         previous polling cycle.
    When:
       - Calling finish().
    Then:
       - The rows are parsed back from JSON and returned; the accumulated responses are restored
         onto the instance for verbose output; and all polling context keys are cleared.
    """
    store = _install_fake_context(monkeypatch)

    rows = [
        {
            "Domain": "evil.example.com",
            "Brand": "Panorama",
            "Status": STATUS_DONE,
            "Result": RESULT_SUCCESS,
            "RuleName": "Cortex - Block Domain",
            "Message": "ok",
        }
    ]
    responses = [[{"HumanReadable": "HR", "Contents": "ok", "Type": 1, "Metadata": None}]]
    store["block_domain_rows"] = json.dumps(rows)
    store["panorama_responses"] = json.dumps(responses)
    store["commit_job_id"] = "999"

    pan_os = _pan_os(["evil.example.com"])
    result = pan_os.finish()

    assert result == rows
    assert pan_os.responses == responses
    # All polling context keys are cleared on finish.
    assert store["commit_job_id"] == ""
    assert store["push_job_id"] == ""
    assert store["panorama_responses"] == ""
    assert store["block_domain_rows"] == ""


def test_finish_tolerates_corrupt_context_data(monkeypatch):
    """
    Given:
       - Context holds a non-JSON (corrupt) block_domain_rows value.
    When:
       - Calling finish().
    Then:
       - finish() does not raise; it degrades to an empty list and still clears context.
    """
    store = _install_fake_context(monkeypatch)
    store["block_domain_rows"] = "{not valid json"
    store["panorama_responses"] = "also not json"

    pan_os = _pan_os(["evil.example.com"])
    result = pan_os.finish()

    assert result == []
    assert store["block_domain_rows"] == ""