44604f1ce3
pytest-based test suite with fixtures for database testing. Covers misc.py utilities, dbs.py operations, and fetch.py validation. Includes mock_network.py for future network testing.
249 lines
8.5 KiB
Python
249 lines
8.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for misc.py utility functions."""
|
|
from __future__ import print_function
|
|
|
|
import re
|
|
import sys
|
|
import os
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import misc
|
|
|
|
|
|
class TestTimestamp:
|
|
"""Tests for timestamp() function."""
|
|
|
|
def test_timestamp_format(self):
|
|
"""timestamp() returns HH:MM:SS format."""
|
|
ts = misc.timestamp()
|
|
assert re.match(r'^\d{2}:\d{2}:\d{2}$', ts)
|
|
|
|
def test_timestamp_valid_hours(self):
|
|
"""timestamp() hours are 00-23."""
|
|
ts = misc.timestamp()
|
|
hours = int(ts.split(':')[0])
|
|
assert 0 <= hours <= 23
|
|
|
|
def test_timestamp_valid_minutes(self):
|
|
"""timestamp() minutes are 00-59."""
|
|
ts = misc.timestamp()
|
|
minutes = int(ts.split(':')[1])
|
|
assert 0 <= minutes <= 59
|
|
|
|
def test_timestamp_valid_seconds(self):
|
|
"""timestamp() seconds are 00-59."""
|
|
ts = misc.timestamp()
|
|
seconds = int(ts.split(':')[2])
|
|
assert 0 <= seconds <= 59
|
|
|
|
|
|
class TestTorProxyUrl:
|
|
"""Tests for tor_proxy_url() function."""
|
|
|
|
def test_basic_format(self):
|
|
"""tor_proxy_url() returns socks5:// prefix."""
|
|
result = misc.tor_proxy_url('127.0.0.1:9050')
|
|
assert result == 'socks5://127.0.0.1:9050'
|
|
|
|
def test_custom_host_port(self):
|
|
"""tor_proxy_url() works with custom host:port."""
|
|
result = misc.tor_proxy_url('10.200.1.1:9150')
|
|
assert result == 'socks5://10.200.1.1:9150'
|
|
|
|
def test_ipv6_host(self):
|
|
"""tor_proxy_url() handles IPv6 addresses."""
|
|
result = misc.tor_proxy_url('[::1]:9050')
|
|
assert result == 'socks5://[::1]:9050'
|
|
|
|
|
|
class TestLogLevel:
|
|
"""Tests for set_log_level() and get_log_level() functions."""
|
|
|
|
def setup_method(self):
|
|
"""Reset log level before each test."""
|
|
misc.set_log_level(1) # Default info level
|
|
|
|
def test_get_default_level(self):
|
|
"""get_log_level() returns default level."""
|
|
assert misc.get_log_level() == 1
|
|
|
|
def test_set_integer_level(self):
|
|
"""set_log_level() accepts integer."""
|
|
misc.set_log_level(0)
|
|
assert misc.get_log_level() == 0
|
|
|
|
misc.set_log_level(2)
|
|
assert misc.get_log_level() == 2
|
|
|
|
def test_set_string_debug(self):
|
|
"""set_log_level('debug') sets level 0."""
|
|
misc.set_log_level('debug')
|
|
assert misc.get_log_level() == 0
|
|
|
|
def test_set_string_info(self):
|
|
"""set_log_level('info') sets level 1."""
|
|
misc.set_log_level('info')
|
|
assert misc.get_log_level() == 1
|
|
|
|
def test_set_string_warn(self):
|
|
"""set_log_level('warn') sets level 2."""
|
|
misc.set_log_level('warn')
|
|
assert misc.get_log_level() == 2
|
|
|
|
def test_set_string_error(self):
|
|
"""set_log_level('error') sets level 3."""
|
|
misc.set_log_level('error')
|
|
assert misc.get_log_level() == 3
|
|
|
|
def test_set_string_none(self):
|
|
"""set_log_level('none') suppresses all."""
|
|
misc.set_log_level('none')
|
|
assert misc.get_log_level() == 99
|
|
|
|
def test_invalid_string_defaults_to_info(self):
|
|
"""set_log_level() with invalid string defaults to 1."""
|
|
misc.set_log_level('invalid')
|
|
assert misc.get_log_level() == 1
|
|
|
|
|
|
class TestIsSSLProtocolError:
|
|
"""Tests for is_ssl_protocol_error() function."""
|
|
|
|
def test_none_returns_false(self):
|
|
"""is_ssl_protocol_error(None) returns False."""
|
|
assert misc.is_ssl_protocol_error(None) is False
|
|
|
|
def test_empty_returns_false(self):
|
|
"""is_ssl_protocol_error('') returns False."""
|
|
assert misc.is_ssl_protocol_error('') is False
|
|
|
|
def test_wrong_version_number(self):
|
|
"""Detects 'wrong version number' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('wrong version number') is True
|
|
|
|
def test_unsupported_protocol(self):
|
|
"""Detects 'unsupported protocol' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('unsupported protocol') is True
|
|
|
|
def test_unexpected_eof(self):
|
|
"""Detects 'unexpected eof' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('unexpected eof') is True
|
|
|
|
def test_eof_occurred(self):
|
|
"""Detects 'eof occurred' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('eof occurred') is True
|
|
|
|
def test_alert_handshake_failure(self):
|
|
"""Detects 'alert handshake failure' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('alert handshake failure') is True
|
|
|
|
def test_http_request(self):
|
|
"""Detects 'http request' as protocol error (sent HTTP to HTTPS)."""
|
|
assert misc.is_ssl_protocol_error('http request') is True
|
|
|
|
def test_no_ciphers_available(self):
|
|
"""Detects 'no ciphers available' as protocol error."""
|
|
assert misc.is_ssl_protocol_error('no ciphers available') is True
|
|
|
|
def test_case_insensitive(self):
|
|
"""is_ssl_protocol_error() is case-insensitive."""
|
|
assert misc.is_ssl_protocol_error('WRONG VERSION NUMBER') is True
|
|
assert misc.is_ssl_protocol_error('Wrong Version Number') is True
|
|
|
|
def test_certificate_error_not_protocol(self):
|
|
"""Certificate errors are not protocol errors."""
|
|
assert misc.is_ssl_protocol_error('certificate verify failed') is False
|
|
|
|
def test_hostname_mismatch_not_protocol(self):
|
|
"""Hostname mismatch is not protocol error."""
|
|
assert misc.is_ssl_protocol_error('hostname mismatch') is False
|
|
|
|
def test_expired_cert_not_protocol(self):
|
|
"""Expired certificate is not protocol error."""
|
|
assert misc.is_ssl_protocol_error('certificate has expired') is False
|
|
|
|
def test_embedded_in_message(self):
|
|
"""Detects pattern embedded in longer message."""
|
|
assert misc.is_ssl_protocol_error(
|
|
'SSL error: wrong version number in record') is True
|
|
|
|
|
|
class TestFailureConstants:
|
|
"""Tests for failure category constants."""
|
|
|
|
def test_constants_are_strings(self):
|
|
"""Failure constants are strings."""
|
|
assert isinstance(misc.FAIL_TIMEOUT, str)
|
|
assert isinstance(misc.FAIL_REFUSED, str)
|
|
assert isinstance(misc.FAIL_AUTH, str)
|
|
assert isinstance(misc.FAIL_UNREACHABLE, str)
|
|
assert isinstance(misc.FAIL_DNS, str)
|
|
assert isinstance(misc.FAIL_SSL, str)
|
|
assert isinstance(misc.FAIL_CLOSED, str)
|
|
assert isinstance(misc.FAIL_PROXY, str)
|
|
assert isinstance(misc.FAIL_OTHER, str)
|
|
|
|
def test_constants_unique(self):
|
|
"""Failure constants have unique values."""
|
|
constants = [
|
|
misc.FAIL_TIMEOUT,
|
|
misc.FAIL_REFUSED,
|
|
misc.FAIL_AUTH,
|
|
misc.FAIL_UNREACHABLE,
|
|
misc.FAIL_DNS,
|
|
misc.FAIL_SSL,
|
|
misc.FAIL_CLOSED,
|
|
misc.FAIL_PROXY,
|
|
misc.FAIL_OTHER,
|
|
]
|
|
assert len(constants) == len(set(constants))
|
|
|
|
def test_ssl_errors_contains_ssl(self):
|
|
"""SSL_ERRORS contains FAIL_SSL."""
|
|
assert misc.FAIL_SSL in misc.SSL_ERRORS
|
|
|
|
def test_conn_errors_contents(self):
|
|
"""CONN_ERRORS contains connection-related failures."""
|
|
assert misc.FAIL_TIMEOUT in misc.CONN_ERRORS
|
|
assert misc.FAIL_REFUSED in misc.CONN_ERRORS
|
|
assert misc.FAIL_UNREACHABLE in misc.CONN_ERRORS
|
|
assert misc.FAIL_CLOSED in misc.CONN_ERRORS
|
|
assert misc.FAIL_DNS in misc.CONN_ERRORS
|
|
# Auth and proxy errors are not connection errors
|
|
assert misc.FAIL_AUTH not in misc.CONN_ERRORS
|
|
assert misc.FAIL_PROXY not in misc.CONN_ERRORS
|
|
|
|
|
|
class TestLogLevels:
|
|
"""Tests for LOG_LEVELS dictionary."""
|
|
|
|
def test_debug_is_lowest(self):
|
|
"""debug is the lowest (most verbose) level."""
|
|
assert misc.LOG_LEVELS['debug'] == 0
|
|
|
|
def test_info_is_one(self):
|
|
"""info level is 1."""
|
|
assert misc.LOG_LEVELS['info'] == 1
|
|
|
|
def test_warn_is_two(self):
|
|
"""warn level is 2."""
|
|
assert misc.LOG_LEVELS['warn'] == 2
|
|
|
|
def test_error_is_three(self):
|
|
"""error level is 3."""
|
|
assert misc.LOG_LEVELS['error'] == 3
|
|
|
|
def test_none_suppresses_all(self):
|
|
"""none level (99) suppresses all output."""
|
|
assert misc.LOG_LEVELS['none'] == 99
|
|
|
|
def test_aliases_equal_info(self):
|
|
"""rate, scraper, stats, diag are aliases for info."""
|
|
assert misc.LOG_LEVELS['rate'] == misc.LOG_LEVELS['info']
|
|
assert misc.LOG_LEVELS['scraper'] == misc.LOG_LEVELS['info']
|
|
assert misc.LOG_LEVELS['stats'] == misc.LOG_LEVELS['info']
|
|
assert misc.LOG_LEVELS['diag'] == misc.LOG_LEVELS['info']
|