test_database.py

"""
Liveness / integration tests for Hash-Buster online services.

Each test submits a known hash of the common password "password" to a service
and checks that the correct plaintext comes back. A FAIL means that service (or
that hash type on that service) is currently NOT working -- it is down, changed
its API/HTML, rate-limited us, or the hash is no longer in its database.

These tests hit the real network. Run only the liveness suite with:

    python -m pytest test_database.py -v
    python -m pytest test_database.py -v -m network      # explicit
    python -m pytest test_database.py -v -m "not network" # skip (offline)
"""
import concurrent.futures
import time

import pytest

from database import Database

# Known hashes of the plaintext "password" for each supported algorithm.
HASHES = {
    'md5':    '5f4dcc3b5aa765d61d8327deb882cf99',
    'ntlm':   '8846f7eaee8fb117ad06bdd830b7586c',
    'sha1':   '5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8',
    'sha256': '5e884898da28047151d0e56f8dc6292773603d0d6aabbdd62a11ef721d1542d8',
    'sha384': 'a8b64babd0aca91a59bdbb7761b421d4f2bb38280d3a75ba0f21f2bebc45583d4'
              '46c598660c94ce680c47d19c30783a7',
    'sha512': 'b109f3bbbc244eb82441917ed06d618b9008dd09b3befd1b5e07394c706a8bb98'
              '0b1d7785e5976ec049b46df5f1326af5a2ea6d103fd07c95385ffab0cacbc86',
}
EXPECTED = 'password'
CALL_TIMEOUT = 20  # seconds; a service that hangs longer counts as down
RETRY_DELAY = 2    # seconds; retry once after a delay to ride out rate limits

# Each service -> the hash types it claims to support (docstring / mapping).
# Covers both the original 5 services and the 2 new ones so the suite reports
# liveness for every website in the project.
SERVICES = {
    Database.md5decrypt: ['md5', 'ntlm', 'sha1', 'sha256', 'sha384', 'sha512'],
    Database.gromweb:    ['md5', 'sha1'],
    Database.md5hashing: ['md5', 'sha1', 'sha256', 'sha384', 'sha512'],
    Database.crackcrypt: ['md5', 'sha1', 'sha256', 'sha512'],
    Database.weakpass:   ['md5', 'ntlm', 'sha1', 'sha256'],
}

# Flatten into (service, hashtype) test cases.
CASES = [(svc, ht) for svc, hts in SERVICES.items() for ht in hts]


def _call(service, hashvalue, hashtype):
    """Run a service call under a hard timeout so a dead host can't hang pytest."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(service, hashvalue, hashtype)
        return future.result(timeout=CALL_TIMEOUT)


@pytest.mark.network
@pytest.mark.parametrize(
    'service,hashtype',
    CASES,
    ids=[f'{svc.__name__}-{ht}' for svc, ht in CASES],
)
def test_service_liveness(service, hashtype):
    name = service.__name__
    result = None
    # Retry once on empty/error: some services (e.g. crackcrypt: 1 req/s) rate-limit us.
    for attempt in range(2):
        try:
            result = _call(service, HASHES[hashtype], hashtype)
        except concurrent.futures.TimeoutError:
            if attempt == 0:
                time.sleep(RETRY_DELAY)
                continue
            pytest.fail(f'{name} ({hashtype}): DOWN - no response within {CALL_TIMEOUT}s')
        except Exception as exc:  # network error, bad JSON, etc.
            if attempt == 0:
                time.sleep(RETRY_DELAY)
                continue
            pytest.fail(f'{name} ({hashtype}): DOWN - {type(exc).__name__}: {exc}')
        if result:
            break
        if attempt == 0:
            time.sleep(RETRY_DELAY)

    # Services return False (or empty) when they cannot crack / are broken.
    if not result:
        pytest.fail(f'{name} ({hashtype}): returned no plaintext (not working / hash not in DB)')

    # Some services return a list from re.findall; normalise before comparing.
    if isinstance(result, (list, tuple)):
        result = result[0] if result else ''

    assert str(result).strip() == EXPECTED, (
        f'{name} ({hashtype}): expected {EXPECTED!r}, got {result!r}'
    )