diff --git a/tests/test_p2pool_stats.py b/tests/test_p2pool_stats.py index ecae184..dfdea9f 100644 --- a/tests/test_p2pool_stats.py +++ b/tests/test_p2pool_stats.py @@ -1 +1,157 @@ -PENDING \ No newline at end of file +# SPDX-License-Identifier: MIT-0 +"""Run with: python3 -m unittest discover -s tests""" +import http.client +import json +import os +import sys +import tempfile +import threading +import time +import unittest +from http.server import ThreadingHTTPServer + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "files")) +import p2pool_stats # noqa: E402 + +STRATUM = { + "hashrate_15m": 8123, "hashrate_1h": 7990, "hashrate_24h": 8011, + "total_hashes": 123456789, "shares_found": 42, "shares_failed": 1, + "average_effort": 97.5, "current_effort": 12.25, "connections": 3, + "incoming_connections": 3, "block_reward_share_percent": 0.42, + "last_share_found_time": 1794090000, + "workers": [ + "203.0.113.7:51234,4000,123,40000,rig1", + "198.51.100.9:40000,3000,456,40000,laptop", + "192.0.2.1:3333,1123,789,40000,48mFvxYDSCzM", + ], +} +POOL = {"pool_list": ["pplns"], "pool_statistics": { + "hashRate": 25000000, "miners": 512, "totalHashes": 10 ** 15, + "lastBlockFoundTime": 1794000000, "lastBlockFound": 3801000, + "totalBlocksFound": 321, "pplnsWeight": 1, "pplnsWindowSize": 2160, + "sidechainDifficulty": 160000000, "sidechainHeight": 999999}} +NETWORK = {"difficulty": 859250000000, "hash": "ab" * 32, "height": 3801526, + "reward": 609800000000, "timestamp": 1794098000} + + +def write(root, rel, data, mtime=None): + path = os.path.join(root, *rel.split("/")) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as handle: + handle.write(data if isinstance(data, str) else json.dumps(data)) + if mtime is not None: + os.utime(path, (mtime, mtime)) + + +class SummarizeTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def full(self): + write(self.root, "local/stratum", STRATUM) + write(self.root, "pool/stats", POOL) + write(self.root, "network/stats", NETWORK) + + def test_full_summary(self): + self.full() + s = p2pool_stats.summarize(self.root, "nano") + self.assertEqual(s["sidechain"], "nano") + self.assertEqual(s["node"]["hashrate_15m"], 8123) + self.assertEqual(s["node"]["miners"], 3) + self.assertEqual(s["node"]["reward_share_percent"], 0.42) + self.assertEqual(s["pool"]["hashrate"], 25000000) + self.assertEqual(s["pool"]["difficulty"], 160000000) + self.assertAlmostEqual(s["network"]["hashrate"], 859250000000 / 120) + self.assertAlmostEqual(s["network"]["reward_xmr"], 0.6098) + + def test_never_leaks_worker_addresses_or_wallets(self): + self.full() + body = json.dumps(p2pool_stats.summarize(self.root, "nano")) + for secret in ("203.0.113.7", "198.51.100.9", "192.0.2.1", "rig1", "48mFvx", "workers", "ab" * 32): + self.assertNotIn(secret, body) + + def test_missing_files_yield_nulls(self): + s = p2pool_stats.summarize(self.root, "mini") + self.assertIsNone(s["node"]) + self.assertIsNone(s["pool"]) + self.assertIsNone(s["network"]) + + def test_malformed_files_are_ignored(self): + write(self.root, "local/stratum", "{not json") + write(self.root, "pool/stats", {"pool_statistics": "nope"}) + write(self.root, "network/stats", [1, 2, 3]) + s = p2pool_stats.summarize(self.root, "nano") + self.assertEqual((s["node"], s["pool"], s["network"]), (None, None, None)) + + def test_old_files_are_flagged_stale(self): + self.full() + old = int(time.time() - p2pool_stats.STALE_AFTER - 60) + write(self.root, "local/stratum", STRATUM, mtime=old) + s = p2pool_stats.summarize(self.root, "nano") + self.assertTrue(s["node"]["stale"]) + self.assertEqual(s["node"]["updated_at"], old) + self.assertFalse(s["network"]["stale"]) + + def test_oversized_file_is_ignored(self): + write(self.root, "local/stratum", " " * (p2pool_stats.MAX_FILE_BYTES + 1)) + self.assertIsNone(p2pool_stats.summarize(self.root, "nano")["node"]) + + def test_non_numeric_fields_become_null(self): + bad = dict(STRATUM, hashrate_15m="9999", hashrate_1h=True, hashrate_24h=-5, + shares_found=float("nan")) + del bad["workers"] + write(self.root, "local/stratum", bad) + node = p2pool_stats.summarize(self.root, "nano")["node"] + self.assertEqual([node[k] for k in ("hashrate_15m", "hashrate_1h", "hashrate_24h", "shares_found")], + [None, None, None, None]) + self.assertEqual(node["miners"], 3) # falls back to connection count + + +class ServerTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + cache = p2pool_stats.Cache(self.tmp.name, "nano") + handler = p2pool_stats.make_handler(cache, "/p2pool/stats.json") + self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.cache = cache + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.tmp.cleanup() + + def request(self, path, method="GET"): + conn = http.client.HTTPConnection("127.0.0.1", self.server.server_address[1], timeout=5) + conn.request(method, path) + response = conn.getresponse() + return response.status, dict(response.getheaders()), response.read() + + def test_unavailable_until_p2pool_writes_stats(self): + status, headers, body = self.request("/p2pool/stats.json") + self.assertEqual(status, 503) + self.assertEqual(headers["Content-Type"], "application/json") + self.assertIsNone(json.loads(body)["node"]) + + def test_serves_summary_with_safe_headers(self): + write(self.tmp.name, "local/stratum", STRATUM) + status, headers, body = self.request("/p2pool/stats.json?x=1") + self.assertEqual(status, 200) + self.assertEqual(headers["X-Content-Type-Options"], "nosniff") + self.assertEqual(json.loads(body)["node"]["hashrate_15m"], 8123) + status, headers, body = self.request("/p2pool/stats.json", "HEAD") + self.assertEqual((status, body), (200, b"")) + + def test_only_one_path_is_served(self): + write(self.tmp.name, "local/stratum", STRATUM) + for path in ("/", "/local/stratum", "/p2pool/../local/stratum", "/p2pool/stats.jsonx"): + self.assertEqual(self.request(path)[0], 404, path) + self.assertEqual(self.request("/healthz")[0], 200) + + +if __name__ == "__main__": + unittest.main()