diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_cli.py | 167 |
1 files changed, 167 insertions, 0 deletions
diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..fc4a6e7 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,167 @@ +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +"""Dispatch tests for the command line: exit codes and wiring, not prompts.""" + +import io +import pathlib +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout + +from abusectl import cli + +FIXTURES = pathlib.Path(__file__).parent / "fixtures" + + +class TestInitNonInteractive(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self._tmp.name) + self.config = self.root / "config.toml" + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, *args): + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + code = cli.main(list(args)) + return code, out.getvalue(), err.getvalue() + + def test_it_writes_a_config_from_flags_alone(self): + code, _, _ = self._run( + "--config", str(self.config), "init", + "--non-interactive", "--trusted-relays", "192.0.2.0/24", + ) + self.assertEqual(code, 0) + self.assertIn("192.0.2.0/24", self.config.read_text()) + + def test_a_missing_required_flag_fails_rather_than_prompting(self): + # An agent cannot answer a prompt, so this must not block. + code, _, err = self._run( + "--config", str(self.config), "init", "--non-interactive" + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("--trusted-relays", err) + self.assertFalse(self.config.exists()) + + def test_a_provider_name_supplies_the_relays(self): + code, _, _ = self._run( + "--config", str(self.config), "init", + "--non-interactive", "--provider", "gmail", + ) + self.assertEqual(code, 0) + self.assertIn("74.125.0.0/16", self.config.read_text()) + + def test_an_unknown_provider_is_an_error(self): + code, _, err = self._run( + "--config", str(self.config), "init", + "--non-interactive", "--provider", "nosuchprovider", + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("nosuchprovider", err) + + def test_an_existing_config_is_refused_without_force(self): + self.config.write_text("[general]\n", encoding="utf-8") + code, _, err = self._run( + "--config", str(self.config), "init", + "--non-interactive", "--trusted-relays", "192.0.2.0/24", + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("--force", err) + + def test_force_overwrites_an_existing_config(self): + self.config.write_text('[general]\ntrusted_relays = ["10.0.0.0/8"]\n', + encoding="utf-8") + code, _, _ = self._run( + "--config", str(self.config), "init", + "--non-interactive", "--trusted-relays", "192.0.2.0/24", "--force", + ) + self.assertEqual(code, 0) + self.assertIn("192.0.2.0/24", self.config.read_text()) + + +class TestParse(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self._tmp.name) + self.config = self.root / "config.toml" + self.cases = self.root / "cases" + self.config.write_text( + f'[general]\ntrusted_relays = ["192.0.2.0/24"]\n' + f'cases = "{self.cases}"\n', + encoding="utf-8", + ) + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, *args): + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + code = cli.main(list(args)) + return code, out.getvalue(), err.getvalue() + + def test_it_creates_a_case_and_prints_its_path(self): + code, out, _ = self._run( + "--config", str(self.config), "parse", + str(FIXTURES / "forged-chain.eml"), + ) + self.assertEqual(code, 0) + created = pathlib.Path(out.strip()) + self.assertTrue(created.is_dir()) + self.assertTrue((created / "manifest.json").is_file()) + self.assertTrue((created / "source.eml").is_file()) + + def test_the_manifest_holds_the_iocs_and_auth_verdicts(self): + import json + + _, out, _ = self._run( + "--config", str(self.config), "parse", + str(FIXTURES / "simple.eml"), + ) + manifest = json.loads( + (pathlib.Path(out.strip()) / "manifest.json").read_text() + ) + values = [i["value"] for i in manifest["iocs"]] + self.assertIn("203.0.113.42", values) + self.assertEqual(manifest["auth"]["spf"], "fail") + + def test_no_recipient_address_reaches_the_manifest(self): + _, out, _ = self._run( + "--config", str(self.config), "parse", + str(FIXTURES / "simple.eml"), + ) + text = (pathlib.Path(out.strip()) / "manifest.json").read_text() + self.assertNotIn("example.org", text) + + def test_a_missing_config_points_at_init(self): + code, _, err = self._run( + "--config", str(self.root / "absent.toml"), "parse", + str(FIXTURES / "simple.eml"), + ) + self.assertEqual(code, cli.EXIT_NOT_CONFIGURED) + self.assertIn("abusectl init", err) + + def test_a_missing_message_is_an_error_not_a_traceback(self): + code, _, err = self._run( + "--config", str(self.config), "parse", str(self.root / "absent.eml"), + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("absent.eml", err) + + +if __name__ == "__main__": + unittest.main() |