Add scraper proxy toggle

worker/src/main.py

@@ -4,46 +4,31 @@ import logging
 import os
 import signal
 import sys
+from collections import defaultdict
 from contextlib import suppress
 from datetime import datetime, timedelta, timezone
-from typing import Any
-from urllib.parse import quote
 
 import aiohttp.web as web
 import asyncpg
+import httpx
 from dotenv import load_dotenv
-from telegram import Update
 from telegram.ext import Application, ExtBot
 from telegram.request import HTTPXRequest
 
 from db import close_pool, get_pool
 from health import create_health_app, record_scheduler_run, set_start_time, set_telegram_polling
-from scraper import extract_ad_fields, fetch_ads, get_network_status
+from scraper import extract_ad_fields, fetch_ads
 from notifier import log_notification, notify_new_ad, notify_price_drop, is_user_muted, buffer_for_digest
-from settings import get_turbo_mode
+from settings import get_proxy_enabled, get_turbo_mode, proxy_available
 
 logger = logging.getLogger(__name__)
 
 load_dotenv()
 
 
-def _get_proxy_url() -> str | None:
-    raw = (os.getenv("HTTPS_PROXY") or os.getenv("https_proxy") or "").strip()
-    if not raw:
-        return None
-
-    if "://" in raw:
-        return raw
-
-    parts = raw.split(":", 3)
-    if len(parts) != 4:
-        logger.warning(
-            "Ignoring invalid HTTPS_PROXY value (expected host:port:user:pass or full URL)"
-        )
-        return None
-
-    host, port, username, password = parts
-    return f"http://{quote(username, safe='')}:{quote(password, safe='')}@{host}:{port}"
+class DirectHTTPXRequest(HTTPXRequest):
+    def _build_client(self) -> httpx.AsyncClient:
+        return httpx.AsyncClient(**self._client_kwargs, trust_env=False)
 
 
 def _ad_passes_filters(fields: dict, kw_row: dict) -> bool:
@@ -207,12 +192,13 @@ async def flush_digests(pool: asyncpg.Pool, bot: ExtBot) -> int:
 
 async def scheduler_task(pool: object, bot: ExtBot) -> None:
     while True:
-        proxy_enabled = bool(get_network_status().get("proxy_enabled"))
+        proxy_enabled = False
         turbo_mode = False
         try:
+            proxy_enabled = proxy_available() and await get_proxy_enabled(pool)
             turbo_mode = await get_turbo_mode(pool)
         except Exception:
-            logger.exception("Could not load turbo mode setting")
+            logger.exception("Could not load scheduler mode settings")
 
         turbo_active = turbo_mode and proxy_enabled
         speed_divisor = 10 if turbo_active else 1
@@ -449,25 +435,23 @@ async def main() -> None:
     except Exception:
         logger.exception("Failed to start Web UI (optional)")
 
-    proxy_url = _get_proxy_url()
-    bot_request = HTTPXRequest(
+    bot_request = DirectHTTPXRequest(
         connection_pool_size=10,
-        proxy_url=proxy_url,
         read_timeout=30.0,
         write_timeout=30.0,
         connect_timeout=30.0,
         pool_timeout=10.0,
         media_write_timeout=60.0,
     )
-    updates_request = HTTPXRequest(
+    updates_request = DirectHTTPXRequest(
         connection_pool_size=10,
-        proxy_url=proxy_url,
         read_timeout=30.0,
         write_timeout=30.0,
         connect_timeout=30.0,
         pool_timeout=10.0,
         media_write_timeout=60.0,
     )
+    logger.info("Telegram Bot API traffic uses direct network path")
 
     app = (
         Application.builder()

worker/src/migrations/08-proxy-setting.sql

@@ -0,0 +1,3 @@
+INSERT INTO app_settings (key, value)
+VALUES ('proxy_enabled', 'true')
+ON CONFLICT (key) DO NOTHING;

worker/src/scraper.py

@@ -1,38 +1,36 @@
 import asyncio
 import logging
 import os
-from datetime import datetime, timezone
+from datetime import datetime
 from typing import Any
-from urllib.parse import quote
 
 import httpx
 
+from db import get_pool
+from settings import get_effective_proxy_url, get_proxy_url_from_env, proxy_available
+
 logger = logging.getLogger(__name__)
 
 _client: httpx.AsyncClient | None = None
+_client_proxy_url: str | None = None
 _proxy_ip_logged = False
 _system_public_ip: str | None = None
 _proxy_public_ip: str | None = None
 _last_used_public_ip: str | None = None
+_proxy_enabled_effective: bool | None = None
 
 
 def _get_proxy_url() -> str | None:
-    raw = (os.getenv("HTTPS_PROXY") or os.getenv("https_proxy") or "").strip()
-    if not raw:
-        return None
+    return get_proxy_url_from_env()
 
-    if "://" in raw:
-        return raw
 
-    parts = raw.split(":", 3)
-    if len(parts) != 4:
-        logger.warning(
-            "Ignoring invalid HTTPS_PROXY value (expected host:port:user:pass or full URL)"
-        )
-        return None
+async def _get_effective_proxy_url() -> str | None:
+    global _proxy_enabled_effective
 
-    host, port, username, password = parts
-    return f"http://{quote(username, safe='')}:{quote(password, safe='')}@{host}:{port}"
+    pool = await get_pool()
+    proxy_url = await get_effective_proxy_url(pool)
+    _proxy_enabled_effective = proxy_url is not None
+    return proxy_url
 
 
 def _redact_proxy_url(proxy_url: str) -> str:
@@ -88,12 +86,22 @@ async def _log_proxy_ip_comparison(proxy_url: str) -> None:
 
 async def get_client() -> httpx.AsyncClient:
     """Return a shared AsyncClient with keepalive connection pool."""
-    global _client, _system_public_ip, _last_used_public_ip
+    global _client, _client_proxy_url, _system_public_ip, _last_used_public_ip
+
+    proxy_url = await _get_effective_proxy_url()
+
+    if _client is not None and not _client.is_closed and _client_proxy_url != proxy_url:
+        await _client.aclose()
+        logger.info(
+            "Recreated httpx client because proxy changed: %s -> %s",
+            "enabled" if _client_proxy_url else "disabled",
+            "enabled" if proxy_url else "disabled",
+        )
+        _client = None
 
     if _client is None or _client.is_closed:
         max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10"))
         max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5"))
-        proxy_url = _get_proxy_url()
 
         if proxy_url:
             await _log_proxy_ip_comparison(proxy_url)
@@ -115,6 +123,7 @@ async def get_client() -> httpx.AsyncClient:
             proxy=proxy_url,
             trust_env=False,
         )
+        _client_proxy_url = proxy_url
         logger.info(
             "Created httpx client: max_conns=%d, keepalive=%d, proxy=%s",
             max_conns,
@@ -127,9 +136,10 @@ async def get_client() -> httpx.AsyncClient:
 
 async def refresh_network_status() -> dict[str, str | bool | None]:
     """Ensure network status has best-effort values even before first scrape."""
-    global _system_public_ip, _proxy_public_ip, _last_used_public_ip
+    global _system_public_ip, _proxy_public_ip, _last_used_public_ip, _proxy_enabled_effective
 
-    proxy_url = _get_proxy_url()
+    proxy_url = await _get_effective_proxy_url()
+    _proxy_enabled_effective = proxy_url is not None
 
     if _system_public_ip is None:
         direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
@@ -155,12 +165,14 @@ async def refresh_network_status() -> dict[str, str | bool | None]:
 
 def get_network_status() -> dict[str, str | bool | None]:
     """Return best-effort network status for UI display."""
-    proxy_enabled = bool(_get_proxy_url())
+    available = proxy_available()
+    proxy_enabled = bool(_proxy_enabled_effective) if _proxy_enabled_effective is not None else available
     last_used = _last_used_public_ip
     if last_used is None:
         last_used = _proxy_public_ip if proxy_enabled else _system_public_ip
 
     return {
+        "proxy_available": available,
         "proxy_enabled": proxy_enabled,
         "system_public_ip": _system_public_ip,
         "proxy_public_ip": _proxy_public_ip,
@@ -170,11 +182,12 @@ def get_network_status() -> dict[str, str | bool | None]:
 
 async def close_client() -> None:
     """Close the shared AsyncClient. Call during shutdown."""
-    global _client
+    global _client, _client_proxy_url
     if _client and not _client.is_closed:
         await _client.aclose()
         logger.info("Closed httpx client")
-        _client = None
+    _client = None
+    _client_proxy_url = None
 
 
 _API_URL = (

worker/src/settings.py

@@ -1,5 +1,42 @@
+import logging
+import os
+from urllib.parse import quote
+
 import asyncpg
 
+logger = logging.getLogger(__name__)
+
+TRUE_VALUES = {"1", "true", "yes", "on"}
+
+
+def _as_bool(raw: object, default: bool = False) -> bool:
+    if raw is None:
+        return default
+    return str(raw).lower() in TRUE_VALUES
+
+
+def get_proxy_url_from_env() -> str | None:
+    raw = (os.getenv("HTTPS_PROXY") or os.getenv("https_proxy") or "").strip()
+    if not raw:
+        return None
+
+    if "://" in raw:
+        return raw
+
+    parts = raw.split(":", 3)
+    if len(parts) != 4:
+        logger.warning(
+            "Ignoring invalid HTTPS_PROXY value (expected host:port:user:pass or full URL)"
+        )
+        return None
+
+    host, port, username, password = parts
+    return f"http://{quote(username, safe='')}:{quote(password, safe='')}@{host}:{port}"
+
+
+def proxy_available() -> bool:
+    return get_proxy_url_from_env() is not None
+
 
 async def ensure_app_settings(pool: asyncpg.Pool) -> None:
     await pool.execute(
@@ -13,20 +50,43 @@ async def ensure_app_settings(pool: asyncpg.Pool) -> None:
     )
 
 
-async def get_turbo_mode(pool: asyncpg.Pool) -> bool:
+async def get_bool_setting(pool: asyncpg.Pool, key: str, default: bool = False) -> bool:
     await ensure_app_settings(pool)
-    raw = await pool.fetchval("SELECT value FROM app_settings WHERE key = 'turbo_mode'")
-    return str(raw).lower() in {"1", "true", "yes", "on"}
+    raw = await pool.fetchval("SELECT value FROM app_settings WHERE key = $1", key)
+    return _as_bool(raw, default=default)
 
 
-async def set_turbo_mode(pool: asyncpg.Pool, enabled: bool) -> None:
+async def set_bool_setting(pool: asyncpg.Pool, key: str, enabled: bool) -> None:
     await ensure_app_settings(pool)
     await pool.execute(
         """
         INSERT INTO app_settings (key, value, updated_at)
-        VALUES ('turbo_mode', $1, now())
+        VALUES ($1, $2, now())
         ON CONFLICT (key)
         DO UPDATE SET value = EXCLUDED.value, updated_at = now()
         """,
+        key,
         "true" if enabled else "false",
     )
+
+
+async def get_turbo_mode(pool: asyncpg.Pool) -> bool:
+    return await get_bool_setting(pool, "turbo_mode")
+
+
+async def set_turbo_mode(pool: asyncpg.Pool, enabled: bool) -> None:
+    await set_bool_setting(pool, "turbo_mode", enabled)
+
+
+async def get_proxy_enabled(pool: asyncpg.Pool) -> bool:
+    return await get_bool_setting(pool, "proxy_enabled", default=proxy_available())
+
+
+async def set_proxy_enabled(pool: asyncpg.Pool, enabled: bool) -> None:
+    await set_bool_setting(pool, "proxy_enabled", enabled)
+
+
+async def get_effective_proxy_url(pool: asyncpg.Pool) -> str | None:
+    if not await get_proxy_enabled(pool):
+        return None
+    return get_proxy_url_from_env()

worker/src/templates/dashboard.html

@@ -46,14 +46,25 @@
         <div class="stat-subtext muted">Worker polling rhythm</div>
     </div>
 
-    <div class="card stat-card" style="border-left: 4px solid {% if data.proxy_enabled %}var(--success-text){% else %}var(--border-hover){% endif %};">
+    <div class="card stat-card" style="border-left: 4px solid {% if data.proxy_enabled %}var(--success-text){% elif data.proxy_available %}var(--warning-text){% else %}var(--border-hover){% endif %};">
         <div class="stat-label">HTTP Proxy</div>
-        <div class="stat-value" style="font-size: 24px; color: {% if data.proxy_enabled %}var(--success-text){% else %}var(--text-primary){% endif %};">
-            {{ 'Enabled' if data.proxy_enabled else 'Disabled' }}
+        <div class="stat-value" style="font-size: 24px; color: {% if data.proxy_enabled %}var(--success-text){% elif data.proxy_available %}var(--warning-text){% else %}var(--text-primary){% endif %};">
+            {% if data.proxy_enabled %}Active{% elif data.proxy_available %}Bypassed{% else %}Unavailable{% endif %}
         </div>
         <div class="stat-subtext muted">
-            {{ 'Outbound scraping uses proxy' if data.proxy_enabled else 'Direct connection' }}
+            {% if data.proxy_enabled %}
+            Scraper uses proxy
+            {% elif data.proxy_available %}
+            Proxy present, scraper direct
+            {% else %}
+            No proxy in .env
+            {% endif %}
         </div>
+        <form method="post" action="/admin/proxy" style="margin-top: 12px;">
+            <button type="submit" {% if not data.proxy_available %}disabled{% endif %} style="border: 1px solid var(--border-hover); background: {% if data.proxy_enabled %}var(--warning-bg){% else %}var(--bg-surface-active){% endif %}; color: var(--text-primary); border-radius: 8px; padding: 8px 12px; cursor: {% if data.proxy_available %}pointer{% else %}not-allowed{% endif %}; font-weight: 600; opacity: {% if data.proxy_available %}1{% else %}0.55{% endif %};">
+                {{ 'Disable Proxy' if data.proxy_enabled else 'Enable Proxy' }}
+            </button>
+        </form>
     </div>
 
     <div class="card stat-card" style="border-left: 4px solid var(--brand-primary);">
@@ -91,7 +102,7 @@
             {% if data.proxy_enabled %}
             Scheduler runs every ~{{ '%.1f'|format(data.turbo_effective_sleep_s) }}s
             {% else %}
-            Enable proxy first to activate turbo
+            Activate proxy first to enable turbo
             {% endif %}
         </div>
         {% if data.seconds_until_next is not none %}

worker/src/web.py

@@ -1,18 +1,16 @@
-import os
 import logging
 import time
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
 
 import httpx
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
-import asyncpg
 
 from db import get_pool
 from scraper import get_network_status, refresh_network_status
-from settings import get_turbo_mode, set_turbo_mode
+from settings import get_proxy_enabled, get_turbo_mode, proxy_available, set_proxy_enabled, set_turbo_mode
 
 logger = logging.getLogger(__name__)
 
@@ -81,10 +79,6 @@ def format_postcodes(postcodes: list | None) -> str:
     return ", ".join(str(p) for p in postcodes)
 
 
-def _proxy_enabled() -> bool:
-    return bool((os.getenv("HTTPS_PROXY") or os.getenv("https_proxy") or "").strip())
-
-
 def _flag_from_country_code(country_code: str | None) -> str:
     if not country_code or len(country_code) != 2:
         return ""
@@ -162,6 +156,9 @@ async def dashboard(request: Request):
             {"request": request, "error": "Database unavailable", "data": None},
         )
 
+    proxy_configured = proxy_available()
+    proxy_setting_enabled = await get_proxy_enabled(pool)
+
     try:
         network = await refresh_network_status()
     except Exception:
@@ -169,7 +166,7 @@ async def dashboard(request: Request):
         network = get_network_status()
     system_ip = network.get("system_public_ip")
     last_used_ip = network.get("last_used_public_ip")
-    proxy_enabled = bool(network.get("proxy_enabled"))
+    proxy_enabled = proxy_configured and proxy_setting_enabled
     system_country = await _country_code_for_ip(system_ip if isinstance(system_ip, str) else None)
     last_used_country = await _country_code_for_ip(last_used_ip if isinstance(last_used_ip, str) else None)
     turbo_mode = await get_turbo_mode(pool)
@@ -199,7 +196,9 @@ async def dashboard(request: Request):
         "queue_pending": queue_pending or 0,
         "queue_dead": queue_dead or 0,
         "last_scheduler": last_scheduler["scraped_at"] if last_scheduler else None,
+        "proxy_available": proxy_configured,
         "proxy_enabled": proxy_enabled,
+        "proxy_setting_enabled": proxy_setting_enabled,
         "system_public_ip": system_ip,
         "last_used_public_ip": last_used_ip,
         "system_country_code": system_country,
@@ -222,6 +221,17 @@ async def toggle_turbo(request: Request):
     return RedirectResponse(url="/", status_code=303)
 
 
+@app.post("/admin/proxy")
+async def toggle_proxy(request: Request):
+    if not proxy_available():
+        return RedirectResponse(url="/", status_code=303)
+
+    pool = await get_pool()
+    current = await get_proxy_enabled(pool)
+    await set_proxy_enabled(pool, not current)
+    return RedirectResponse(url="/", status_code=303)
+
+
 @app.get("/keywords", response_class=HTMLResponse)
 async def keywords_list(request: Request):
     try:
@@ -384,4 +394,4 @@ async def stats_json():
         "queue_failed": queue_failed or 0,
         "digest_buffered": digest_buffered or 0,
         "last_scheduler_run": last_scheduler["scraped_at"].isoformat() if last_scheduler else None,
-    }
+    }

worker/tests/test_proxy_settings.py

@@ -0,0 +1,81 @@
+"""Tests for the app-level proxy setting."""
+
+import pytest
+
+from settings import get_proxy_enabled, set_proxy_enabled
+
+
+def _clear_proxy_env(monkeypatch):
+    monkeypatch.delenv("HTTPS_PROXY", raising=False)
+    monkeypatch.delenv("https_proxy", raising=False)
+
+
+@pytest.mark.asyncio
+async def test_get_proxy_enabled_defaults_true_when_https_proxy_is_usable(
+    mock_pool, monkeypatch
+):
+    """First-run default should enable only with a usable proxy."""
+    _clear_proxy_env(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
+    mock_pool.fetchval.return_value = None
+
+    assert await get_proxy_enabled(mock_pool) is True
+
+    mock_pool.fetchval.assert_awaited_once()
+    assert "proxy_enabled" in mock_pool.fetchval.await_args.args
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("proxy_value", [None, "", "proxy.example:8080:user"])
+async def test_get_proxy_enabled_defaults_false_without_usable_https_proxy(
+    mock_pool, monkeypatch, proxy_value
+):
+    """Missing, blank, and malformed proxy env values default disabled."""
+    _clear_proxy_env(monkeypatch)
+    if proxy_value is not None:
+        monkeypatch.setenv("HTTPS_PROXY", proxy_value)
+    mock_pool.fetchval.return_value = None
+
+    assert await get_proxy_enabled(mock_pool) is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("stored_value", "expected"),
+    [
+        ("true", True),
+        ("1", True),
+        ("yes", True),
+        ("on", True),
+        ("false", False),
+        ("0", False),
+        ("no", False),
+        ("off", False),
+    ],
+)
+async def test_get_proxy_enabled_uses_persisted_setting(
+    mock_pool, monkeypatch, stored_value, expected
+):
+    """Once saved, the DB setting should control the UI/runtime toggle."""
+    _clear_proxy_env(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
+    mock_pool.fetchval.return_value = stored_value
+
+    assert await get_proxy_enabled(mock_pool) is expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("enabled", "stored_value"),
+    [(True, "true"), (False, "false")],
+)
+async def test_set_proxy_enabled_persists_proxy_toggle(
+    mock_pool, enabled, stored_value
+):
+    """The proxy toggle should be stored under the app setting key."""
+    await set_proxy_enabled(mock_pool, enabled)
+
+    mock_pool.execute.assert_awaited()
+    sql, *args = mock_pool.execute.await_args.args
+    assert "proxy_enabled" in sql or "proxy_enabled" in args
+    assert stored_value in args

worker/tests/test_scraper_proxy.py

@@ -0,0 +1,129 @@
+"""Proxy parsing and runtime-status tests for scraper."""
+
+import inspect
+
+import pytest
+
+import scraper
+
+
+def _clear_proxy_env(monkeypatch):
+    monkeypatch.delenv("HTTPS_PROXY", raising=False)
+    monkeypatch.delenv("https_proxy", raising=False)
+
+
+def _proxy_url_parser():
+    parser = getattr(scraper, "_get_proxy_url", None)
+    if parser is None:
+        pytest.skip("scraper does not expose a proxy URL parser")
+    return parser
+
+
+async def _get_network_status(mock_pool, monkeypatch):
+    monkeypatch.setattr(
+        scraper,
+        "_proxy_enabled_effective",
+        None,
+        raising=False,
+    )
+    status_fn = getattr(scraper, "get_network_status")
+    signature = inspect.signature(status_fn)
+    if "pool" in signature.parameters:
+        result = status_fn(mock_pool)
+        if inspect.isawaitable(result):
+            return await result
+        return result
+
+    effective_proxy_fn = getattr(scraper, "_get_effective_proxy_url", None)
+    if effective_proxy_fn is not None:
+        async def fake_get_pool():
+            return mock_pool
+
+        monkeypatch.setattr(scraper, "get_pool", fake_get_pool, raising=False)
+        result = effective_proxy_fn()
+        if inspect.isawaitable(result):
+            await result
+
+    result = status_fn()
+    if inspect.isawaitable(result):
+        return await result
+    return result
+
+
+def _block_public_ip_fetch(monkeypatch):
+    async def fail_fetch(*_args, **_kwargs):
+        raise AssertionError("network status tests must not fetch public IPs")
+
+    monkeypatch.setattr(scraper, "_fetch_public_ip", fail_fetch, raising=False)
+
+
+def test_proxy_url_parser_accepts_full_proxy_url(monkeypatch):
+    _clear_proxy_env(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "http://user:pass@proxy.example:8080")
+
+    assert _proxy_url_parser()() == "http://user:pass@proxy.example:8080"
+
+
+def test_proxy_url_parser_builds_url_from_colon_format(monkeypatch):
+    _clear_proxy_env(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user@example:p a/s")
+
+    assert (
+        _proxy_url_parser()()
+        == "http://user%40example:p%20a%2Fs@proxy.example:8080"
+    )
+
+
+@pytest.mark.parametrize("proxy_value", ["", "proxy.example:8080:user"])
+def test_proxy_url_parser_rejects_missing_or_malformed_proxy(
+    monkeypatch, proxy_value
+):
+    _clear_proxy_env(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", proxy_value)
+
+    assert _proxy_url_parser()() is None
+
+
+@pytest.mark.asyncio
+async def test_network_status_reports_proxy_available_when_env_is_usable(
+    mock_pool, monkeypatch
+):
+    _clear_proxy_env(monkeypatch)
+    _block_public_ip_fetch(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
+    mock_pool.fetchval.return_value = None
+
+    status = await _get_network_status(mock_pool, monkeypatch)
+
+    assert status["proxy_available"] is True
+    assert status["proxy_enabled"] is True
+
+
+@pytest.mark.asyncio
+async def test_network_status_respects_disabled_proxy_setting(
+    mock_pool, monkeypatch
+):
+    _clear_proxy_env(monkeypatch)
+    _block_public_ip_fetch(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
+    mock_pool.fetchval.return_value = "false"
+
+    status = await _get_network_status(mock_pool, monkeypatch)
+
+    assert status["proxy_available"] is True
+    assert status["proxy_enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_network_status_reports_unavailable_for_malformed_proxy(
+    mock_pool, monkeypatch
+):
+    _clear_proxy_env(monkeypatch)
+    _block_public_ip_fetch(monkeypatch)
+    monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user")
+    mock_pool.fetchval.return_value = None
+
+    status = await _get_network_status(mock_pool, monkeypatch)
+
+    assert status["proxy_available"] is False
+    assert status["proxy_enabled"] is False