ansible-role-monero/tests/test_p2pool_stats.py

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