# 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()