B2 - uh

Author: Omena0

tests/test_core.py

@@ -0,0 +1,81 @@
+from typing import Any, Dict
+import httplib as app
+import pathlib
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_url_encode_decode() -> None:
+    s = 'Hello World!'
+    encoded = await app.url_encode(s)
+    assert '%20' in encoded
+    decoded = await app.url_decode(encoded)
+    assert decoded == s
+
+@pytest.mark.asyncio
+async def test_get_content_type() -> None:
+    assert await app.get_content_type('index.html') == 'text/html'
+    assert await app.get_content_type('style.css') == 'text/css'
+    assert await app.get_content_type('data.json') == 'application/json'
+    assert await app.get_content_type('image.png') == 'image/png'
+
+def test_cookie_set_delete_headers():
+    resp = app.Response()
+    resp.set_cookie('x', 'y', path='/', max_age=3600, httponly=True, secure=True, samesite='Lax')
+    headers = resp.get_cookie_headers()
+    assert any('x=y' in h for h in headers)
+    resp.delete_cookie('x')
+    headers = resp.get_cookie_headers()
+    assert any('x=' in h and 'Expires' in h or 'expires' in h for h in headers)
+
+def test_user_create_login_session(tmp_path: pathlib.Path) -> None:
+    # Clear state
+    app.users.clear()
+    app.profiles.clear()
+    app.sessions.clear()
+
+    assert app.create_user('alice', 'pass123') is True
+    assert app.user_exists('alice') is True
+    assert app.login('alice', 'pass123') is True
+    assert app.login('alice', 'wrong') is False
+
+    token = app.start_session('alice')
+    assert app.verify_session(token) is True
+    assert app.get_username(token) == 'alice'
+    app.end_session(token)
+    assert app.verify_session(token) is False
+
+def test_save_load_users(tmp_path: pathlib.Path) -> None:
+    # Prepare users
+    app.users.clear()
+    app.profiles.clear()
+    app.create_user('bob', 'secret')
+    app.update_user_profile('bob', 'hi')
+
+    p = tmp_path / 'users.json'
+    assert app.save_users(str(p)) is True
+    # Clear and load
+    app.users.clear()
+    app.profiles.clear()
+    assert app.load_users(str(p)) is True
+    assert app.user_exists('bob')
+    profile: Dict[str, Any] | None = app.get_user_profile('bob')
+    assert profile is not None
+    assert profile['bio'] == 'hi'
+
+@pytest.mark.asyncio
+async def test_template_and_redirect(tmp_path: pathlib.Path, monkeypatch: 'pytest.MonkeyPatch') -> None:
+    # create a static file
+    f = tmp_path / 'index.html'
+    f.write_text('<h1>{{ title }}</h1>')
+    # monkeypatch os.path to use tmp_path
+    monkeypatch.setenv('PYHTTPLIB_TEST_STATIC', str(tmp_path))
+    # patch _normalize_path to return our file name
+    async def fake_normalize(path: str) -> str:
+        return str(f)
+    monkeypatch.setattr(app, '_normalize_path', fake_normalize)
+    out = await app.template('index.html', title='Hi')
+    assert '<h1>Hi</h1>' in out
+    code, html = await app.redirect('https://example.com')
+    assert code == 302
+    assert 'refresh' in html.lower()

tests/test_more_core.py

@@ -0,0 +1,113 @@
+import pytest
+import httplib as app
+import os
+import socket
+import types
+
+@pytest.mark.asyncio
+async def test_normalize_path_variants():
+    assert await app._normalize_path('/') == 'static/index.html'
+    assert await app._normalize_path('/about') == 'static/about.html'
+    assert await app._normalize_path('/static/page.html') == 'static/page.html'
+    # traversal should default to index
+    assert await app._normalize_path('../etc/passwd') == 'static/index.html'
+
+@pytest.mark.asyncio
+async def test_parsers():
+    headers = 'Host: example.com\nX-Test: value\n'
+    parsed = await app._parse_headers(headers)
+    assert parsed['Host'] == 'example.com'
+    assert parsed['X-Test'] == 'value'
+
+    params = await app._parse_params('a=1&b=two')
+    assert params['a'] == '1' and params['b'] == 'two'
+
+    cookies = await app._parse_cookies('a=1; b=2')
+    assert cookies['a'] == '1' and cookies['b'] == '2'
+
+@pytest.mark.asyncio
+async def test_error_shorthand_and_content_type():
+    assert await app.get_error_shorthand(200) == 'OK'
+    assert await app.get_error_shorthand(418) == "I'm a teapot"
+    assert await app.get_error_shorthand(999) == 'Unknown Error'
+
+    assert await app.get_content_type('index.html') == 'text/html'
+    assert await app.get_content_type('file.unknown') == 'text/plain'
+
+def test_client_ip_and_request_context(monkeypatch):
+    # prepare request context
+    req = app._get_request()
+    req.client_ip = '1.2.3.4'
+    assert app.get_client_ip() == '1.2.3.4'
+    req.client_ip = ''
+    req.headers['X-Forwarded-For'] = '5.6.7.8, 9.9.9.9'
+    assert app.get_client_ip() == '5.6.7.8'
+    req.headers.clear()
+
+def test_dos_and_blocking():
+    ip = '10.0.0.1'
+    # ensure clean
+    if ip in app.dos_ip_requests: del app.dos_ip_requests[ip]
+    if ip in app.dos_blocked_ips: del app.dos_blocked_ips[ip]
+
+    # simulate many requests quickly to trigger block
+    for _ in range(app.DOS_MAX_REQUESTS + 1):
+        blocked = app.check_dos_protection(ip)
+    assert app.is_ip_blocked(ip) is True
+    blocked_ips = app.get_blocked_ips()
+    assert ip in blocked_ips
+    # unblock
+    assert app.unblock_ip(ip) is True
+    assert app.is_ip_blocked(ip) is False
+
+def test_dos_stats_empty():
+    stats = app.get_dos_stats()
+    assert 'tracked_ips' in stats and 'blocked_ips' in stats
+
+@pytest.mark.asyncio
+async def test_match_route_and_build_response():
+    vars = await app._match_route('/user/123', '/user/<id>')
+    assert vars == {'id': '123'}
+    assert await app._match_route('/a/b', '/a/b') == {}
+    assert await app._match_route('/a/b', '/a/c') is None
+
+    # test response building with cookies
+    resp = app._get_response()
+    resp.set_cookie('s', 'v', max_age=10, httponly=True)
+    header = await app._build_response(200, 'text/plain')
+    assert b'Set-Cookie' in header
+
+def test_user_operations_and_sessions(tmp_path):
+    app.users.clear(); app.profiles.clear(); app.sessions.clear()
+    assert app.create_user('x', 'p') is True
+    assert app.user_exists('x')
+    assert app.login('x', 'p') is True
+    assert app.edit_user('x', 'new') is True
+    assert app.delete_user('x') is True
+
+    app.create_user('y', 'p')
+    token = app.start_session('y')
+    assert app.verify_session(token)
+    assert app.get_username(token) == 'y'
+    app.end_session(token)
+    assert not app.verify_session(token)
+
+@pytest.mark.asyncio
+async def test_ratelimit_decorator_and_socket_mode(monkeypatch):
+    # Decorator mode: ensure it returns a callable
+    dec = app.ratelimit(0.1)
+    @dec
+    async def f():
+        return 'ok'
+    assert callable(f)
+
+    # Socket mode: create a fake socket with getpeername
+    class FakeSock:
+        def getpeername(self):
+            return ('127.0.0.1', 12345)
+    sock = FakeSock()
+    # delay-based: first call allowed, immediate second call should be rate-limited
+    assert app.ratelimit(sock, 0.5, 'test') is False
+    assert app.ratelimit(sock, 0.5, 'test') is True
+
+*** End Patch

tests/test_ws_and_misc.py

@@ -0,0 +1,95 @@
+import os
+import time
+import pytest
+import httplib as app
+import socket
+from typing import Any
+from pathlib import Path
+
+
+class FakeSock(socket.socket):
+    def __init__(self, ip: str = '127.0.0.1'):
+        super().__init__(socket.AF_INET, socket.SOCK_STREAM)
+        self._ip = ip
+
+    def getpeername(self):
+        return (self._ip, 12345)
+
+
+def setup_function(func: Any) -> None:
+    # Reset DOS and rate-limit related global state before each test
+    app.ws_dos_violations.clear()
+    app.ws_last_message.clear()
+    app.ws_messages_per_second.clear()
+    app.dos_ip_requests.clear()
+    app.dos_blocked_ips.clear()
+    app.ws_rate_limit_delays.clear()
+    app.ws_rate_limit_counts.clear()
+    app.rate_limit_delays.clear()
+    app.rate_limit_counts.clear()
+
+
+def test_check_websocket_dos_disconnect():
+    sock = FakeSock('9.9.9.9')
+
+    # Ensure fresh state
+    if sock in app.ws_dos_violations: del app.ws_dos_violations[sock]
+    if sock in app.ws_messages_per_second: del app.ws_messages_per_second[sock]
+    if sock in app.ws_last_message: del app.ws_last_message[sock]
+
+    now = time.time()
+    # Simulate message count already at the limit
+    app.ws_messages_per_second[sock] = (now, app.DOS_WS_MAX_MESSAGES_PER_SECOND)
+    app.ws_dos_violations[sock] = app.DOS_WS_MAX_VIOLATIONS - 1
+
+    # This should increment violations and cause a disconnect (True)
+    res = app._check_websocket_dos_protection(sock)
+    assert res is True
+    assert app.ws_dos_violations.get(sock, 0) >= app.DOS_WS_MAX_VIOLATIONS
+
+
+def test_ratelimit_count_based_socket_mode():
+    sock = FakeSock('127.0.0.1')
+    key = (sock, 'count_test')
+    # Use count-based: allow 2 requests per 1 second
+    # First two should be allowed, third should be rate-limited
+    assert app.ratelimit(sock, (2, 1), 'count_test') is False
+    assert app.ratelimit(sock, (2, 1), 'count_test') is False
+    assert app.ratelimit(sock, (2, 1), 'count_test') is True
+
+
+@pytest.mark.asyncio
+async def test_template_redirect_and_save_load_users(tmp_path: Path):
+    # Create static dir and a simple template file
+    os.makedirs('static', exist_ok=True)
+    tpl_path = os.path.join('static', 'test_template.html')
+    with open(tpl_path, 'w', encoding='utf-8') as f:
+        f.write('<h1>{{ name }}</h1>')
+
+    # Render template
+    out = await app.template('test_template.html', name='Alice')
+    assert '<h1>Alice</h1>' in out
+
+    # Redirect
+    code, html = await app.redirect('https://example.com')
+    assert code == 302
+    assert 'https://example.com' in html
+
+    # Save and load users
+    app.users.clear(); app.profiles.clear()
+    assert app.create_user('u1', 'pass') is True
+    tmpfile = tmp_path / 'users.json'
+    assert app.save_users(str(tmpfile)) is True
+
+    # Clear and reload
+    app.users.clear(); app.profiles.clear()
+    assert app.load_users(str(tmpfile)) is True
+    assert 'u1' in app.users
+
+    # Cleanup files created
+    try:
+        os.remove(tpl_path)
+        if not os.listdir('static'):
+            os.rmdir('static')
+    except Exception:
+        pass