test: add Docker E2E and live connector test files

Adds test_docker_full.py (4 docker-marked tests against a running stack),
test_live_keboola.py, test_live_bigquery.py, and test_live_jira.py (live-marked,
read-only, skipped when credentials are absent).
This commit is contained in:
ZdenekSrotyr 2026-04-12 11:10:06 +02:00
parent c24205a1bf
commit 4d8de9c3b7
4 changed files with 215 additions and 0 deletions

93
tests/test_docker_full.py Normal file
View file

@ -0,0 +1,93 @@
"""Docker E2E tests — requires a running docker-compose stack.
Run with: pytest tests/test_docker_full.py -m docker -v
Assumes docker compose is already up and healthy at DOCKER_TEST_URL.
"""
import os
import time
import httpx
import pytest
pytestmark = pytest.mark.docker
DOCKER_BASE_URL = os.environ.get("DOCKER_TEST_URL", "http://localhost:8000")
def _wait_for_healthy(url: str, timeout: int = 60) -> bool:
"""Poll GET /api/health until 200 or timeout."""
deadline = time.time() + timeout
while time.time() < deadline:
try:
resp = httpx.get(f"{url}/api/health", timeout=5)
if resp.status_code == 200:
return True
except Exception:
pass
time.sleep(1)
return False
@pytest.fixture(scope="module", autouse=True)
def require_docker():
"""Wait for the docker stack to be healthy before running tests."""
if not _wait_for_healthy(DOCKER_BASE_URL, timeout=60):
pytest.skip(
f"Docker stack at {DOCKER_BASE_URL} did not become healthy within 60s. "
"Start it with: docker compose up"
)
def test_app_health():
"""Health endpoint returns 200 with status and version fields."""
resp = httpx.get(f"{DOCKER_BASE_URL}/api/health", timeout=10)
assert resp.status_code == 200
data = resp.json()
assert "status" in data
assert "version" in data
def test_app_returns_html_on_root():
"""GET / returns 200 (HTML dashboard)."""
resp = httpx.get(f"{DOCKER_BASE_URL}/", timeout=10)
assert resp.status_code == 200
def test_bootstrap_creates_admin():
"""POST /auth/bootstrap creates the first admin user (409 if already done)."""
resp = httpx.post(
f"{DOCKER_BASE_URL}/auth/bootstrap",
json={"email": "admin@docker-test.local", "name": "Docker Admin", "password": "test1234"},
timeout=10,
)
# 200 = created, 409 = already bootstrapped — both are valid
assert resp.status_code in (200, 409)
def test_trigger_sync():
"""Login then POST /api/sync/trigger returns accepted."""
# First bootstrap or login to get a token
bootstrap = httpx.post(
f"{DOCKER_BASE_URL}/auth/bootstrap",
json={"email": "admin@docker-test.local", "name": "Docker Admin", "password": "test1234"},
timeout=10,
)
if bootstrap.status_code == 200:
token = bootstrap.json()["access_token"]
else:
# Already bootstrapped — log in
login = httpx.post(
f"{DOCKER_BASE_URL}/auth/token",
json={"email": "admin@docker-test.local", "password": "test1234"},
timeout=10,
)
if login.status_code != 200:
pytest.skip("Cannot obtain admin token — adjust credentials or bootstrap the stack")
token = login.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
resp = httpx.post(f"{DOCKER_BASE_URL}/api/sync/trigger", headers=headers, timeout=15)
# 200 = started, 202 = accepted/queued
assert resp.status_code in (200, 202)

View file

@ -0,0 +1,37 @@
"""Live BigQuery tests — require real GCP credentials in environment variables.
Run with: pytest tests/test_live_bigquery.py -m live -v
Requires: BIGQUERY_PROJECT, GOOGLE_APPLICATION_CREDENTIALS environment variables.
All tests are read-only; no data is written or deleted.
"""
import os
import pytest
pytestmark = pytest.mark.live
BIGQUERY_PROJECT = os.environ.get("BIGQUERY_PROJECT", "")
GOOGLE_APPLICATION_CREDENTIALS = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "")
@pytest.fixture(autouse=True)
def require_bigquery_env():
"""Skip all tests in this module if BigQuery credentials are missing."""
if not BIGQUERY_PROJECT or not GOOGLE_APPLICATION_CREDENTIALS:
pytest.skip(
"BigQuery credentials not set. "
"Export BIGQUERY_PROJECT and GOOGLE_APPLICATION_CREDENTIALS to run live tests."
)
def test_simple_query():
"""BigQuery client can execute a trivial SELECT 1 query."""
from google.cloud import bigquery
client = bigquery.Client(project=BIGQUERY_PROJECT)
query_job = client.query("SELECT 1 as x")
rows = list(query_job.result())
assert len(rows) == 1
assert rows[0]["x"] == 1

37
tests/test_live_jira.py Normal file
View file

@ -0,0 +1,37 @@
"""Live Jira tests — require real Jira credentials in environment variables.
Run with: pytest tests/test_live_jira.py -m live -v
Requires: JIRA_DOMAIN, JIRA_EMAIL, JIRA_API_TOKEN environment variables.
All tests are read-only; no data is written or deleted.
"""
import os
import httpx
import pytest
pytestmark = pytest.mark.live
JIRA_DOMAIN = os.environ.get("JIRA_DOMAIN", "")
JIRA_EMAIL = os.environ.get("JIRA_EMAIL", "")
JIRA_API_TOKEN = os.environ.get("JIRA_API_TOKEN", "")
@pytest.fixture(autouse=True)
def require_jira_env():
"""Skip all tests in this module if Jira credentials are missing."""
if not JIRA_DOMAIN or not JIRA_EMAIL or not JIRA_API_TOKEN:
pytest.skip(
"Jira credentials not set. "
"Export JIRA_DOMAIN, JIRA_EMAIL, and JIRA_API_TOKEN to run live tests."
)
def test_jira_myself():
"""Jira /rest/api/3/myself returns 200 with valid credentials."""
url = f"https://{JIRA_DOMAIN}/rest/api/3/myself"
resp = httpx.get(url, auth=(JIRA_EMAIL, JIRA_API_TOKEN), timeout=15)
assert resp.status_code == 200
data = resp.json()
assert "accountId" in data or "emailAddress" in data

View file

@ -0,0 +1,48 @@
"""Live Keboola tests — require real credentials in environment variables.
Run with: pytest tests/test_live_keboola.py -m live -v
Requires: KBC_STORAGE_TOKEN, KBC_STACK_URL environment variables.
All tests are read-only; no data is written or deleted.
"""
import os
import pytest
pytestmark = pytest.mark.live
KBC_STORAGE_TOKEN = os.environ.get("KBC_STORAGE_TOKEN", "")
KBC_STACK_URL = os.environ.get("KBC_STACK_URL", "")
@pytest.fixture(autouse=True)
def require_keboola_env():
"""Skip all tests in this module if Keboola credentials are missing."""
if not KBC_STORAGE_TOKEN or not KBC_STACK_URL:
pytest.skip(
"Keboola credentials not set. "
"Export KBC_STORAGE_TOKEN and KBC_STACK_URL to run live tests."
)
def test_connection():
"""KeboolaClient.test_connection() returns True with valid credentials."""
from connectors.keboola.client import KeboolaClient
client = KeboolaClient(token=KBC_STORAGE_TOKEN, url=KBC_STACK_URL)
assert client.test_connection() is True
def test_discover_tables():
"""KeboolaClient.discover_all_tables() returns a non-empty list of tables."""
from connectors.keboola.client import KeboolaClient
client = KeboolaClient(token=KBC_STORAGE_TOKEN, url=KBC_STACK_URL)
tables = client.discover_all_tables()
assert isinstance(tables, list)
assert len(tables) > 0, "Expected at least one table in the Keboola project"
# Verify structure of first table entry
first = tables[0]
assert "id" in first
assert "name" in first