From 734d389b2029c747a5086018978d6b418e741238 Mon Sep 17 00:00:00 2001 From: tmdeveloper007 Date: Wed, 1 Jul 2026 20:58:03 +0000 Subject: [PATCH] test: add unit tests for port-scanner plugin parser --- .../fixtures/port-scanner/sample_output.txt | 9 + .../backend/unit/test_port_scanner_plugin.py | 235 ++++++++++++++++++ 2 files changed, 244 insertions(+) create mode 100644 testing/backend/unit/fixtures/port-scanner/sample_output.txt create mode 100644 testing/backend/unit/test_port_scanner_plugin.py diff --git a/testing/backend/unit/fixtures/port-scanner/sample_output.txt b/testing/backend/unit/fixtures/port-scanner/sample_output.txt new file mode 100644 index 000000000..d71222ce0 --- /dev/null +++ b/testing/backend/unit/fixtures/port-scanner/sample_output.txt @@ -0,0 +1,9 @@ +Starting Nmap scan of 192.168.1.1 +Host is up (0.0010s latency). +PORT STATE SERVICE +22/tcp open ssh +80/tcp open http +443/tcp open https +3306/tcp open mysql +445/tcp open microsoft-ds +8080/tcp open http-proxy diff --git a/testing/backend/unit/test_port_scanner_plugin.py b/testing/backend/unit/test_port_scanner_plugin.py new file mode 100644 index 000000000..e8aeceae6 --- /dev/null +++ b/testing/backend/unit/test_port_scanner_plugin.py @@ -0,0 +1,235 @@ +""" +Unit tests for plugins/port-scanner/parser.py. +""" + +import importlib.util +import sys +from pathlib import Path + +import pytest + +# Import the parser module directly from the file without requiring __init__.py +_parser_path = Path(__file__).resolve().parents[3] / "plugins" / "port-scanner" / "parser.py" +_spec = importlib.util.spec_from_file_location("plugins.port_scanner.parser", str(_parser_path)) +_parser_module = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_parser_module) +parse = _parser_module.parse + + +class TestParseOpenPorts: + def test_finds_single_open_port(self): + output = "22/tcp open ssh" + result = parse(output) + assert result["count"] >= 1 + + def test_finds_multiple_open_ports(self): + output = ( + "22/tcp open ssh\n" + "80/tcp open http\n" + "443/tcp open https\n" + ) + result = parse(output) + open_ports = result["structured"]["rows"] + assert len(open_ports) == 3 + ports = [p["port"] for p in open_ports] + assert "22" in ports + assert "80" in ports + assert "443" in ports + + def test_skips_closed_ports(self): + output = "22/tcp closed ssh\n80/tcp open http" + result = parse(output) + open_ports = result["structured"]["rows"] + assert len(open_ports) == 1 + assert open_ports[0]["port"] == "80" + + def test_skips_filtered_ports(self): + output = "22/tcp filtered firewall\n80/tcp open http" + result = parse(output) + open_ports = result["structured"]["rows"] + assert len(open_ports) == 1 + + def test_port_protocol_in_structured_data(self): + output = "8080/tcp open http-proxy" + result = parse(output) + port = result["structured"]["rows"][0] + assert port["port"] == "8080" + assert port["protocol"] == "tcp" + assert port["service"] == "http-proxy" + + def test_udp_ports_parsed(self): + output = "53/udp open domain" + result = parse(output) + open_ports = result["structured"]["rows"] + assert len(open_ports) == 1 + assert open_ports[0]["protocol"] == "udp" + + +class TestSeverityClassification: + def test_regular_ports_have_low_severity(self): + output = "22/tcp open ssh\n80/tcp open http" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "low" in severities + + def test_mysql_port_triggers_medium(self): + output = "3306/tcp open mysql" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_ftp_port_triggers_medium(self): + output = "21/tcp open ftp" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_telnet_port_triggers_medium(self): + output = "23/tcp open telnet" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_smb_port_triggers_medium(self): + output = "445/tcp open microsoft-ds" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_rdp_port_triggers_medium(self): + output = "3389/tcp open ms-wbt-server" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_postgres_port_triggers_medium(self): + output = "5432/tcp open postgresql" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + def test_multiple_critical_ports_all_medium(self): + output = "21/tcp open ftp\n3306/tcp open mysql\n445/tcp open smb" + result = parse(output) + severities = [f["severity"] for f in result["findings"]] + assert "medium" in severities + + +class TestFindingStructure: + def test_finding_has_correct_title(self): + output = "80/tcp open http" + result = parse(output) + assert result["findings"][0]["title"] == "Open Network Services Detected" + + def test_finding_has_correct_category(self): + output = "22/tcp open ssh" + result = parse(output) + assert result["findings"][0]["category"] == "Insecure Surface" + + def test_finding_has_remediation(self): + output = "80/tcp open http" + result = parse(output) + assert "remediation" in result["findings"][0] + assert len(result["findings"][0]["remediation"]) > 0 + + def test_metadata_contains_open_ports(self): + output = "22/tcp open ssh\n80/tcp open http" + result = parse(output) + meta = result["findings"][0]["metadata"] + assert "open_ports" in meta + assert len(meta["open_ports"]) == 2 + + def test_description_contains_port_count(self): + output = "22/tcp open ssh\n80/tcp open http\n443/tcp open https" + result = parse(output) + desc = result["findings"][0]["description"] + assert "3" in desc + + +class TestStructuredData: + def test_total_count_reflects_open_ports(self): + output = "22/tcp open ssh\n80/tcp open http\n443/tcp open https" + result = parse(output) + assert result["structured"]["total_count"] == 3 + assert result["count"] == 1 + + def test_rows_contain_port_service_protocol(self): + output = "22/tcp open ssh" + result = parse(output) + row = result["structured"]["rows"][0] + assert "port" in row + assert "protocol" in row + assert "service" in row + + def test_type_is_ports(self): + output = "80/tcp open http" + result = parse(output) + assert result["structured"]["type"] == "ports" + + def test_host_status_up(self): + output = "Host is up\n22/tcp open ssh" + result = parse(output) + assert result["structured"]["host_status"] == "up" + + def test_host_status_unknown(self): + output = "22/tcp open ssh" + result = parse(output) + assert result["structured"]["host_status"] == "unknown" + + +class TestSummaryText: + def test_summary_contains_open_port_count(self): + output = "22/tcp open ssh" + result = parse(output) + summary = result["summary"] + assert any("1" in s for s in summary) + + def test_summary_contains_service_name(self): + output = "22/tcp open ssh" + result = parse(output) + summary_text = " ".join(result["summary"]) + assert "ssh" in summary_text.lower() + + def test_empty_summary_when_no_open_ports(self): + output = "22/tcp closed ssh" + result = parse(output) + assert len(result["findings"]) == 0 + + +class TestHostUpFinding: + def test_host_up_no_open_ports_creates_info_finding(self): + output = "Host is up (0.050s latency)" + result = parse(output) + assert len(result["findings"]) == 1 + assert result["findings"][0]["severity"] == "info" + assert result["findings"][0]["title"] == "Host Status: Active" + + def test_host_up_with_closed_ports_no_finding(self): + output = "22/tcp closed ssh\n80/tcp filtered http" + result = parse(output) + assert len(result["findings"]) == 0 + + +class TestEdgeCases: + def test_empty_input_returns_empty_findings(self): + result = parse("") + assert result["count"] == 0 + assert result["findings"] == [] + assert result["structured"]["rows"] == [] + + def test_whitespace_only_input_returns_empty_findings(self): + result = parse(" \n \n ") + assert result["count"] == 0 + assert result["findings"] == [] + + def test_non_matching_lines_skipped(self): + output = "Starting Nmap scan\n22/tcp open ssh\nHost is up\nFinished scan" + result = parse(output) + open_ports = result["structured"]["rows"] + assert len(open_ports) == 1 + assert open_ports[0]["port"] == "22" + + def test_case_insensitive_state_matching(self): + output = "22/tcp OPEN ssh\n80/tcp Open http" + result = parse(output) + assert len(result["structured"]["rows"]) == 2