bifroest/tests/test_sources.py

"""Git-Quellen: Prüfung der URL, Verbindungstest, Chat erst mit getesteter Quelle (→ Konzept 7.3, 7.5)."""

import pytest
from sqlalchemy import select

from app.core.errors import Invalid
from app.data_sources.connectors.git import GitConfig, GitConnector, check_config
from app.data_sources.models import DataSource
from helpers import connect_repo, create_workspace


@pytest.mark.parametrize("url, meldung", [
    ("http://bitbucket.intern/repo.git", "Nur https://"),
    ("https://evil.example/repo.git", "nicht freigegeben"),
    ("https://user:pw@bitbucket.intern/repo.git", "Keine Zugangsdaten"),
    ("ssh://git@bitbucket.intern/repo.git", "Nur https://"),
    ("ext::sh -c touch% /tmp/x", "Nur https://"),
    ("-uhttps://bitbucket.intern/x", "Nur https://"),
])
def test_unzulaessige_urls(settings, url, meldung):
    with pytest.raises(Invalid, match=meldung):
        check_config(url, "main", settings)


def test_erlaubter_host(settings):
    assert check_config("https://bitbucket.intern/scm/p/r.git", "", settings).branch == "main"


def test_file_nur_wenn_erlaubt(settings, git_repo):
    with pytest.raises(Invalid, match="nicht freigegeben"):
        check_config(git_repo, "main", settings.model_copy(update={"git_allow_local": False}))


@pytest.mark.parametrize("branch", ["../x", "a..b", "-rf", "/a", "a/", "x.lock", "a b"])
def test_unzulaessige_branches(settings, branch):
    with pytest.raises(Invalid):
        check_config("https://bitbucket.intern/r.git", branch, settings)


def test_verbindungstest_lokal(settings, git_repo):
    result = GitConnector(GitConfig(git_repo, "main"), settings).test_connection()
    assert result.ok, result.message


def test_branch_fehlt(settings, git_repo):
    result = GitConnector(GitConfig(git_repo, "gibtsnicht"), settings).test_connection()
    assert not result.ok and "nicht gefunden" in result.message


def test_token_nur_in_der_umgebung(settings):
    connector = GitConnector(GitConfig("https://bitbucket.intern/r.git", "main"), settings, token="geheim-123")
    env = connector.environment()
    assert env["GIT_CONFIG_VALUE_0"] == "Authorization: Bearer geheim-123"
    assert env["GIT_ALLOW_PROTOCOL"] == "https:file"
    assert connector.clean("fatal: geheim-123 abgelehnt") == "fatal: *** abgelehnt"


def test_chat_erst_nach_erfolgreichem_test(browser, people, git_repo):
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    assert anna.post(f"/workspaces/{workspace_id}/chats").status_code == 400
    connect_repo(anna, workspace_id, git_repo, branch="gibtsnicht")
    assert anna.post(f"/workspaces/{workspace_id}/chats").status_code == 400
    connect_repo(anna, workspace_id, git_repo)
    response = anna.post(f"/workspaces/{workspace_id}/chats")
    assert response.status_code == 303 and response.headers["location"].startswith("/chats/")


def test_bestaetigung_pflicht(browser, people, git_repo):
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    response = anna.post(f"/workspaces/{workspace_id}/sources", {"name": "R", "url": git_repo})
    assert response.status_code == 400 and "bestätigen" in response.text


def test_token_verschluesselt_und_nur_hinweis_sichtbar(browser, people, db, git_repo, settings):
    from app.core.crypto import Vault
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    anna.post(f"/workspaces/{workspace_id}/sources", {"name": "R", "url": git_repo, "token": "bbdc-geheim-4711",
                                                      "exposure_confirmed": "1"})
    with db.session() as session:
        source = session.scalar(select(DataSource))
    assert "bbdc-geheim" not in source.secret and Vault(settings.secret_key).decrypt(source.secret) == "bbdc-geheim-4711"
    page = anna.get(f"/workspaces/{workspace_id}/data").text
    assert "Token …4711" in page and "bbdc-geheim" not in page
    assert "bbdc-geheim" not in anna.get(f"/sources/{source.id}/edit").text


def test_quelle_bearbeiten_und_entfernen(browser, people, db, git_repo):
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    connect_repo(anna, workspace_id, git_repo)
    with db.session() as session:
        source_id = session.scalar(select(DataSource.id))
    # Token ersetzen: Quelle wird neu getestet (lokales Repo ignoriert das Token)
    response = anna.post(f"/sources/{source_id}", {"name": "Umbenannt", "branch": "main", "token": "neues-token-1234"})
    assert response.status_code == 303
    with db.session() as session:
        source = session.get(DataSource, source_id)
        assert source.name == "Umbenannt" and source.secret_hint == "1234" and source.status == "ok"
    anna.post(f"/sources/{source_id}", {"name": "Umbenannt", "branch": "main", "remove_token": "1"})
    with db.session() as session:
        assert session.get(DataSource, source_id).secret is None
    # Falscher Branch: Fehler beim Test, Chat gesperrt
    anna.post(f"/sources/{source_id}", {"name": "Umbenannt", "branch": "gibtsnicht"})
    with db.session() as session:
        assert session.get(DataSource, source_id).status == "error"
    assert browser("dora").post(f"/sources/{source_id}/delete").status_code == 404
    assert anna.post(f"/sources/{source_id}/delete").status_code == 303
    with db.session() as session:
        assert session.get(DataSource, source_id) is None


def test_nur_admin_bearbeitet_quellen(browser, people, db, git_repo):
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    connect_repo(anna, workspace_id, git_repo)
    anna.post(f"/workspaces/{workspace_id}/members", {"username": "ben", "role": "member"})
    with db.session() as session:
        source_id = session.scalar(select(DataSource.id))
    ben = browser("ben")
    assert ben.get(f"/sources/{source_id}/edit").status_code == 403
    assert ben.post(f"/sources/{source_id}", {"name": "X"}).status_code == 403
    assert ben.post(f"/sources/{source_id}/delete").status_code == 403


def test_testen_per_htmx_liefert_zeile(browser, people, db, git_repo):
    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    connect_repo(anna, workspace_id, git_repo)
    with db.session() as session:
        source_id = session.scalar(select(DataSource.id))
    response = anna.post(f"/sources/{source_id}/test", headers={"HX-Request": "true"})
    assert response.status_code == 200 and f'id="source-{source_id}"' in response.text
    assert browser("dora").post(f"/sources/{source_id}/test").status_code == 404


def test_anmeldung_bearer_oder_nutzername_und_token(browser, people, db, git_repo, settings):
    """Ohne Nutzername Bearer (Bitbucket Data Center), mit Nutzername Basic (Gitea, Forgejo, GitLab, GitHub)."""
    import base64

    import httpx

    from app.core.crypto import Vault
    from app.data_sources import service as sources
    from app.data_sources.connectors import pull_requests
    from app.data_sources.connectors.git import auth_header
    assert auth_header("tok-12345") == "Bearer tok-12345"
    assert auth_header("tok-12345", "bot") == "Basic " + base64.b64encode(b"bot:tok-12345").decode()

    anna = browser("anna")
    workspace_id = create_workspace(anna, people["team_demo"])
    form = {"name": "R", "url": git_repo, "exposure_confirmed": "1"}
    assert "nur zusammen mit einem Token" in anna.post(f"/workspaces/{workspace_id}/sources",
                                                       {**form, "username": "bot"}).text
    assert "ohne Leerzeichen und Doppelpunkt" in anna.post(f"/workspaces/{workspace_id}/sources",
                                                           {**form, "username": "a:b", "token": "tok-12345"}).text
    assert anna.post(f"/workspaces/{workspace_id}/sources",
                     {**form, "username": "bot", "token": "gitea-tok-4711"}).status_code == 303
    vault = Vault(settings.secret_key)
    with db.session() as session:
        source = session.scalar(select(DataSource))
        env = sources.connector(source, settings, vault).environment()
    assert source.config["username"] == "bot" and "gitea-tok" not in str(source.config)
    assert env["GIT_CONFIG_VALUE_0"] == "Authorization: " + auth_header("gitea-tok-4711", "bot")
    assert 'value="bot"' in anna.get(f"/sources/{source.id}/edit").text

    # Nutzername entfernen → wieder Bearer, Quelle wird neu getestet; Token bleibt
    anna.post(f"/sources/{source.id}", {"name": "R", "branch": "main", "username": ""})
    with db.session() as session:
        source = session.get(DataSource, source.id)
        env = sources.connector(source, settings, vault).environment()
    assert env["GIT_CONFIG_VALUE_0"] == "Authorization: Bearer gitea-tok-4711"
    # Token entfernen, Nutzername behalten → abgelehnt
    response = anna.post(f"/sources/{source.id}", {"name": "R", "branch": "main", "username": "bot",
                                                   "remove_token": "1"})
    assert response.status_code == 400 and "nur zusammen mit einem Token" in response.text

    # Die Bitbucket-API bekommt dieselbe Anmeldung
    seen = []
    transport = httpx.MockTransport(lambda r: seen.append(r) or httpx.Response(200, json={"values": []}))
    pull_requests.BitbucketDataCenter("https://bb.intern/scm/P/r.git", "tok-12345", 5, transport=transport,
                                      username="bot").open()
    assert seen[0].headers["Authorization"] == auth_header("tok-12345", "bot")