94 lines
5.7 KiB
Python
94 lines
5.7 KiB
Python
"""DB-Dienst für Oracle: dieselben Tools und Sicherungen wie bei PostgreSQL, mit Oracle-SQL (ab 19c)."""
|
||
|
||
import pytest
|
||
|
||
from tests.conftest import PASSWORD
|
||
|
||
|
||
def text(result) -> str:
|
||
return "".join(c.text for c in result.content if c.type == "text")
|
||
|
||
|
||
def test_tools_wie_bei_postgres(oracle_service):
|
||
tools = {t.name: t for t in oracle_service.call("__list__").tools}
|
||
assert set(tools) == {"check_connection", "list_tables", "describe_table", "query"}
|
||
assert all(t.annotations.read_only_hint for t in tools.values())
|
||
assert "Oracle" in tools["query"].description
|
||
|
||
|
||
def test_verbindung_pruefen(oracle_service, oracle_reader, oracle_database):
|
||
result = oracle_service.call("check_connection", headers=oracle_reader)
|
||
assert not result.is_error, text(result)
|
||
assert text(result).startswith("Oracle 23.") and "2 Tabellen und Views" in text(result)
|
||
assert "Schreibrechte" not in text(result) and "systemweite" not in text(result)
|
||
writer = {**oracle_reader, "X-DB-User": oracle_database["writer"]}
|
||
assert "Schreibrechte auf 1 Objekte" in text(oracle_service.call("check_connection", headers=writer))
|
||
admin = {**oracle_reader, "X-DB-User": oracle_database["admin"], "X-DB-Password": oracle_database["admin_password"]}
|
||
assert "systemweite Rechte (" in text(oracle_service.call("check_connection", headers=admin))
|
||
|
||
|
||
def test_tabellen_und_beschreibung(oracle_service, oracle_reader, oracle_database):
|
||
owner = oracle_database["owner"]
|
||
result = oracle_service.call("list_tables", headers=oracle_reader)
|
||
assert f"- {owner}.KUNDEN (Tabelle, ca. 1200 Zeilen): Alle Kunden" in text(result)
|
||
assert f"- {owner}.TOP (View)" in text(result) and result.structured_content["row_count"] == 2
|
||
# Ohne Anführungszeichen wie in Oracle: Großbuchstaben
|
||
result = oracle_service.call("describe_table", {"table": f"{owner.lower()}.kunden"}, headers=oracle_reader)
|
||
assert not result.is_error, text(result)
|
||
assert "- UMSATZ: NUMBER(10,2) – Umsatz in Euro" in text(result)
|
||
assert "- NAME: VARCHAR2(100) (nie leer)" in text(result) and "- NOTIZ: CLOB\n" in text(result)
|
||
assert result.structured_content["rows"][0] == ["ID", "NUMBER(10)", False, None]
|
||
assert result.structured_content["sample"]["row_count"] == 5
|
||
result = oracle_service.call("describe_table", {"table": "gibt_es_nicht"}, headers=oracle_reader)
|
||
assert result.is_error and f"{oracle_database['reader']}.GIBT_ES_NICHT gibt es nicht" in text(result)
|
||
|
||
|
||
def test_abfrage_mit_tabelle_und_limit(oracle_service, oracle_reader, oracle_database):
|
||
owner = oracle_database["owner"]
|
||
result = oracle_service.call("query", {"sql": f"SELECT id, name, seit, umsatz, notiz FROM {owner}.kunden "
|
||
"ORDER BY id"}, headers=oracle_reader)
|
||
data = result.structured_content
|
||
assert data["columns"] == ["ID", "NAME", "SEIT", "UMSATZ", "NOTIZ"] and data["row_count"] == 100
|
||
assert data["truncated"] and "gekürzt auf 100" in text(result)
|
||
assert data["rows"][0] == [1, "Kunde 1", "2024-01-02T00:00:00", 10.5, "langer Text"]
|
||
result = oracle_service.call("query", {"sql": f"SELECT count(*) AS n, sum(umsatz) AS summe FROM {owner}.kunden;"},
|
||
headers=oracle_reader)
|
||
assert result.structured_content["rows"] == [[1200, 7566300]]
|
||
result = oracle_service.call("query", {"sql": f"WITH t AS (SELECT * FROM {owner}.kunden) SELECT id FROM t "
|
||
"ORDER BY id FETCH FIRST 3 ROWS ONLY"}, headers=oracle_reader)
|
||
assert result.structured_content["rows"] == [[1], [2], [3]]
|
||
|
||
|
||
@pytest.mark.parametrize("sql, message", [
|
||
("DELETE FROM {owner}.kunden", "genau eine lesende Abfrage"),
|
||
("BEGIN DELETE FROM {owner}.kunden; END;", "genau eine lesende Abfrage"),
|
||
("WITH FUNCTION f RETURN NUMBER IS BEGIN RETURN 1; END; SELECT f FROM dual", "PL/SQL"),
|
||
("with /* x */\nfunction f return number is begin return 1; end; select f from dual", "PL/SQL"),
|
||
("SELECT * FROM {owner}.kunden FOR UPDATE", "SQL-Fehler"),
|
||
("SELECT 1 FROM dual; DELETE FROM {owner}.kunden", "SQL-Fehler"),
|
||
("SELECT * FROM gibt_es_nicht", "ORA-00942"),
|
||
])
|
||
def test_schreiben_unmoeglich_auch_mit_schreibrechten(oracle_service, oracle_reader, oracle_database, sql, message):
|
||
writer = {**oracle_reader, "X-DB-User": oracle_database["writer"]}
|
||
result = oracle_service.call("query", {"sql": sql.format(owner=oracle_database["owner"])}, headers=writer)
|
||
assert result.is_error and message in text(result), text(result)
|
||
count = oracle_service.call("query", {"sql": f"SELECT count(*) FROM {oracle_database['owner']}.kunden"},
|
||
headers=writer)
|
||
assert count.structured_content["rows"] == [[1200]]
|
||
|
||
|
||
def test_zeitlimit(oracle_service, oracle_reader):
|
||
result = oracle_service.call("query", {"sql": "SELECT count(*) FROM all_objects a, all_objects b, all_objects c"},
|
||
headers=oracle_reader)
|
||
assert result.is_error and "abgebrochen" in text(result), text(result)
|
||
|
||
|
||
def test_verstaendliche_verbindungsfehler(oracle_service, oracle_reader):
|
||
result = oracle_service.call("check_connection", headers={**oracle_reader, "X-DB-Password": "falsch"})
|
||
assert result.is_error and "Benutzer oder Passwort falsch." in text(result) and "falsch" not in \
|
||
text(result).replace("Passwort falsch", "")
|
||
result = oracle_service.call("check_connection", headers={**oracle_reader, "X-DB-Database": "GIBTS_NICHT"})
|
||
assert result.is_error and "Service-Name unbekannt" in text(result), text(result)
|
||
result = oracle_service.call("check_connection", headers={**oracle_reader, "X-DB-Port": "1"})
|
||
assert result.is_error and "Datenbank nicht erreichbar" in text(result)
|
||
assert PASSWORD not in text(result)
|