"""End-to-end tests: run the real CLI against the fake Graylog server.""" import io import json import os import sys import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) sys.path.insert(0, ROOT) sys.path.insert(0, HERE) from fake_graylog import MESSAGES, FakeGraylog # noqa: E402 from graylog_ops import cli # noqa: E402 class CliEndToEndTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.server = FakeGraylog().start() @classmethod def tearDownClass(cls): cls.server.stop() def setUp(self): self.server.terms_size_param = "size" self.server.requests.clear() def run_cli(self, *argv, json_mode=True): args = ["--url", self.server.url, "--token", "testtoken"] if json_mode: args.append("--json") args.extend(argv) stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): code = cli.main(args) return code, stdout.getvalue(), stderr.getvalue() # ------------------------------------------------------------- probe def test_probe_json(self): code, out, err = self.run_cli("probe") self.assertEqual(code, 0, err) data = json.loads(out) self.assertIn("4.0.8", data["system"]["version"]) self.assertEqual(data["indexer"]["status"], "green") failed = [check["name"] for check in data["checks"] if not check["ok"]] self.assertEqual(failed, [], "unexpected failing endpoints: %s" % failed) def test_probe_human_output(self): code, out, err = self.run_cli("probe", json_mode=False) self.assertEqual(code, 0, err) self.assertIn("检查项", out) self.assertIn("4.0.8", out) self.assertIn("全部", out) # ---------------------------------------------------------- discover def test_discover_finds_client_ip(self): code, out, err = self.run_cli("discover", "--range", "3600") self.assertEqual(code, 0, err) data = json.loads(out) self.assertEqual(data["sampled"], len(MESSAGES)) self.assertEqual(data["recommendation"]["field"], "client_ip") self.assertEqual(data["recommendation"]["confidence"], "high") self.assertNotIn("message", [row["field"] for row in data["candidates"]]) self.assertIn("status", data["http_fields"]) self.assertIn("uri", data["http_fields"]) def test_discover_stream_title_resolution(self): code, out, err = self.run_cli("discover", "--stream", "业务访问") self.assertEqual(code, 0, err) data = json.loads(out) self.assertEqual(data["stream_id"], "5f00000000000000000000aa") def test_discover_unknown_stream_fails_cleanly(self): stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): with self.assertRaises(SystemExit) as ctx: cli.main(["--url", self.server.url, "--token", "testtoken", "discover", "--stream", "does-not-exist"]) self.assertNotEqual(ctx.exception.code, 0) # ------------------------------------------------------------- terms def test_top_ips(self): code, out, err = self.run_cli("top-ips", "--field", "client_ip", "--size", "10") self.assertEqual(code, 0, err) data = json.loads(out) self.assertEqual(data["field"], "client_ip") self.assertEqual(len(data["terms"]), 4) self.assertEqual(data["terms"][0]["term"], "203.0.113.5") self.assertEqual(data["terms"][0]["count"], 3) def test_terms_use_views_backend(self): """4.x removed the legacy terms endpoint, so aggregation must POST.""" code, out, err = self.run_cli("top-ips", "--field", "client_ip") self.assertEqual(code, 0, err) self.assertTrue(json.loads(out)["terms"]) posts = [path for path, _ in self.server.requests if path == "/api/views/search/sync"] self.assertTrue(posts, "expected a views search POST") def test_legacy_fallback_when_views_missing(self): """Graylog 3.x has the legacy terms endpoint but no views API.""" server = FakeGraylog(views_enabled=False).start() server.terms_size_param = "limit" try: stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): code = cli.main(["--url", server.url, "--token", "testtoken", "--json", "terms", "--field", "client_ip"]) self.assertEqual(code, 0, stderr.getvalue()) self.assertTrue(json.loads(stdout.getvalue())["terms"]) term_calls = [query for path, query in server.requests if path.endswith("/universal/terms")] self.assertEqual(len(term_calls), 2) self.assertIn("size", term_calls[0]) self.assertIn("limit", term_calls[1]) self.assertNotIn("size", term_calls[1]) finally: server.stop() # ----------------------------------------------------------- snapshot def test_snapshot(self): with tempfile.TemporaryDirectory() as directory: code, out, err = self.run_cli("snapshot", "--dir", directory, "--sample", "5") self.assertEqual(code, 0, err) data = json.loads(out) failed = [row for row in data["resources"] if row["status"] != "ok"] self.assertEqual(failed, []) self.assertTrue(os.path.exists(os.path.join(directory, "manifest.json"))) self.assertTrue(os.path.exists(os.path.join(directory, "system.json"))) self.assertEqual(data["sampled_messages"], 5) messages_path = os.path.join(directory, "messages.jsonl") with open(messages_path, "r", encoding="utf-8") as handle: lines = [line for line in handle if line.strip()] self.assertEqual(len(lines), 5) json.loads(lines[0]) # -------------------------------------------------------------- raw def test_raw_endpoint(self): code, out, err = self.run_cli("raw", "/api/system/inputs") self.assertEqual(code, 0, err) data = json.loads(out) self.assertIn("inputs", data) # ---------------------------------------------------------- messages def test_messages_jsonl(self): code, out, err = self.run_cli("messages", "--limit", "3", "--jsonl") self.assertEqual(code, 0, err) lines = [line for line in out.splitlines() if line.strip()] self.assertEqual(len(lines), 3) for line in lines: json.loads(line) # --------------------------------------------------------------- auth def test_bad_token_returns_1(self): stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): code = cli.main(["--url", self.server.url, "--token", "wrong", "probe"]) self.assertEqual(code, 1) self.assertIn("401", stderr.getvalue()) def test_missing_config_returns_2(self): saved = {key: os.environ.pop(key, None) for key in ( "GRAYLOG_URL", "GRAYLOG_TOKEN", "GRAYLOG_USER", "GRAYLOG_PASSWORD")} original_cwd = os.getcwd() try: # Run from an empty directory: a real ./graylog-ops.json in the # project root must not leak into this test. with tempfile.TemporaryDirectory() as directory: os.chdir(directory) stderr = io.StringIO() with redirect_stderr(stderr): code = cli.main(["system"]) self.assertEqual(code, 2) self.assertIn("配置错误", stderr.getvalue()) finally: os.chdir(original_cwd) for key, value in saved.items(): if value is not None: os.environ[key] = value # --------------------------------------------------------------- collect def test_collect_bundle(self): with tempfile.TemporaryDirectory() as directory: code, out, err = self.run_cli("collect", "--dir", directory, "--range", "3600") self.assertEqual(code, 0, err) for name in ("probe.json", "discover.json", "judge.json", "profile.json", "SUMMARY.txt"): self.assertTrue(os.path.exists(os.path.join(directory, name)), name) self.assertTrue(os.path.isdir(os.path.join(directory, "snapshot"))) data = json.loads(out) self.assertTrue(all(step["ok"] for step in data["steps"]), data["steps"]) self.assertEqual(data["recommendation"]["field"], "client_ip") # ---------------------------------------------------------------- out def test_out_file(self): with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "ips.json") code, out, err = self.run_cli("top-ips", "--field", "client_ip", "--out", path) self.assertEqual(code, 0, err) self.assertTrue(os.path.exists(path)) with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) self.assertEqual(payload["field"], "client_ip") if __name__ == "__main__": unittest.main()