import time as time_mod from unittest.mock import AsyncMock, MagicMock, patch from urllib.parse import urlparse, parse_qs import pytest from hbd.server import oauth from hbd.server import users as users_mod from hbd.server.users import User CFG_OFF = {} CFG_ON = { "oauth": { "gitea": { "url": "https://git.example.com", "client_id": "cid", "client_secret": "csec", } } } CFG_PARTIAL = {"oauth": {"gitea": {"url": "https://git.example.com"}}} @pytest.fixture(autouse=True) def clear_oauth_states(): oauth._states.clear() yield oauth._states.clear() @pytest.fixture(autouse=True) def reset_users_dict(): original = dict(users_mod.users) yield users_mod.users = original def test_is_enabled_when_all_keys_present(): assert oauth.is_enabled(CFG_ON) is True def test_is_enabled_false_when_no_oauth_key(): assert oauth.is_enabled(CFG_OFF) is False def test_is_enabled_false_when_partial_config(): assert oauth.is_enabled(CFG_PARTIAL) is False def test_make_state_returns_unique_tokens(): s1 = oauth.make_state() s2 = oauth.make_state() assert s1 != s2 assert len(s1) == 64 # 32 bytes hex def test_validate_state_valid(): state = oauth.make_state() assert oauth.validate_state(state) is True def test_validate_state_consumed_on_use(): state = oauth.make_state() oauth.validate_state(state) assert oauth.validate_state(state) is False # replay rejected def test_validate_state_unknown(): assert oauth.validate_state("notastate") is False def test_validate_state_expired(monkeypatch): state = oauth.make_state() # Wind expiry into the past monkeypatch.setitem(oauth._states, state, time_mod.time() - 1000) assert oauth.validate_state(state) is False def _reset_users(entries=None): users_mod.users = entries or {} def test_provision_oauth_user_new(): _reset_users() user = users_mod.provision_oauth_user("gituser", "Git User", "https://example.com/avatar.png") assert user.username == "gituser" assert user.full_name == "Git User" assert user.avatar == "https://example.com/avatar.png" assert user.admin is False assert user.password_hash == "" assert "gituser" in users_mod.users def test_provision_oauth_user_no_password_login(): _reset_users() user = users_mod.provision_oauth_user("gituser", "Git User", "") assert user.check_password("anything") is False def test_provision_oauth_user_existing_updates_profile(): existing = User( username="alice", full_name="Old Name", avatar="old.png", password_hash="pbkdf2:sha256:1:salt:abc", admin=True, notification_channels=["chan1"], ) _reset_users({"alice": existing}) user = users_mod.provision_oauth_user("alice", "New Name", "new.png") assert user.full_name == "New Name" assert user.avatar == "new.png" # Preserved assert user.admin is True assert user.password_hash == "pbkdf2:sha256:1:salt:abc" assert user.notification_channels == ["chan1"] def test_provision_oauth_user_does_not_overwrite_with_empty(): existing = User(username="bob", full_name="Bob", avatar="bob.png") _reset_users({"bob": existing}) user = users_mod.provision_oauth_user("bob", "", "") assert user.full_name == "Bob" assert user.avatar == "bob.png" def test_provision_oauth_user_survives_config_reload(): _reset_users() users_mod.provision_oauth_user("oauthonly", "OAuth Only", "https://example.com/a.png") assert "oauthonly" in users_mod.users # Reload with empty config — OAuth user should survive users_mod.load_users({}) assert "oauthonly" in users_mod.users def test_authorization_url_shape(): state = "teststate" redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" url = oauth.authorization_url(CFG_ON, state, redirect_uri) parsed = urlparse(url) qs = parse_qs(parsed.query) assert parsed.scheme == "https" assert parsed.netloc == "git.example.com" assert parsed.path == "/login/oauth/authorize" assert qs["client_id"] == ["cid"] assert qs["state"] == ["teststate"] assert qs["redirect_uri"] == [redirect_uri] assert qs["scope"] == ["user:email"] assert qs["response_type"] == ["code"] @pytest.mark.asyncio async def test_exchange_code_returns_token(): redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" mock_response = AsyncMock() mock_response.status = 200 mock_response.json = AsyncMock(return_value={"access_token": "tok123"}) mock_session = MagicMock() mock_session.post = MagicMock(return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_response), __aexit__=AsyncMock(return_value=False), )) with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_session), __aexit__=AsyncMock(return_value=False), )): token = await oauth.exchange_code(CFG_ON, "mycode", redirect_uri) assert token == "tok123" @pytest.mark.asyncio async def test_exchange_code_raises_on_error_status(): redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" mock_response = AsyncMock() mock_response.status = 401 mock_response.text = AsyncMock(return_value="unauthorized") mock_session = MagicMock() mock_session.post = MagicMock(return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_response), __aexit__=AsyncMock(return_value=False), )) with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_session), __aexit__=AsyncMock(return_value=False), )): with pytest.raises(oauth.OAuthError): await oauth.exchange_code(CFG_ON, "badcode", redirect_uri) @pytest.mark.asyncio async def test_fetch_user_returns_profile(): mock_response = AsyncMock() mock_response.status = 200 mock_response.json = AsyncMock(return_value={ "login": "alice", "full_name": "Alice Smith", "avatar_url": "https://git.example.com/avatars/alice.png", }) mock_session = MagicMock() mock_session.get = MagicMock(return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_response), __aexit__=AsyncMock(return_value=False), )) with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_session), __aexit__=AsyncMock(return_value=False), )): profile = await oauth.fetch_user(CFG_ON, "tok123") assert profile == { "login": "alice", "full_name": "Alice Smith", "avatar_url": "https://git.example.com/avatars/alice.png", } @pytest.mark.asyncio async def test_exchange_code_raises_when_no_access_token(): redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" mock_response = AsyncMock() mock_response.status = 200 mock_response.json = AsyncMock(return_value={"error": "bad_request"}) mock_session = MagicMock() mock_session.post = MagicMock(return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_response), __aexit__=AsyncMock(return_value=False), )) with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_session), __aexit__=AsyncMock(return_value=False), )): with pytest.raises(oauth.OAuthError): await oauth.exchange_code(CFG_ON, "mycode", redirect_uri) @pytest.mark.asyncio async def test_fetch_user_raises_on_error_status(): mock_response = AsyncMock() mock_response.status = 401 mock_response.text = AsyncMock(return_value="unauthorized") mock_session = MagicMock() mock_session.get = MagicMock(return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_response), __aexit__=AsyncMock(return_value=False), )) with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( __aenter__=AsyncMock(return_value=mock_session), __aexit__=AsyncMock(return_value=False), )): with pytest.raises(oauth.OAuthError): await oauth.fetch_user(CFG_ON, "tok123")