k1nq
e43428c1a2
Gitea Actions CI / detect (push) Successful in 3s
Gitea Actions CI / Compose & nginx validation (push) Failing after 20s
Gitea Actions CI / Frontend (lint / type / test / build) (push) Failing after 24s
Gitea Actions CI / Backend (lint / type / test / build) (push) Failing after 1m34s
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
import os
|
|
import uuid
|
|
from datetime import date
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.exc import DBAPIError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import set_space_context
|
|
from app.imports.blob import get_blob_store
|
|
from app.imports.models import HoldResolution, ParseRun, SourceRecord, StatementDocument
|
|
from app.ledger.models import Account
|
|
from app.spaces.models import FinancialSpace
|
|
|
|
|
|
@pytest.mark.skipif(not os.getenv("RLS_DATABASE_URL"), reason="requires restricted PostgreSQL role")
|
|
def test_postgres_rls_requires_matching_transaction_context() -> None:
|
|
engine = create_engine(os.environ["RLS_DATABASE_URL"])
|
|
with Session(engine) as db:
|
|
space = FinancialSpace(name="RLS integration")
|
|
db.add(space)
|
|
db.commit()
|
|
space_id = space.id
|
|
|
|
db.add(Account(space_id=space_id, name="Blocked", kind="cash", opened_on=date.today()))
|
|
with pytest.raises(DBAPIError):
|
|
db.commit()
|
|
db.rollback()
|
|
|
|
set_space_context(db, space_id)
|
|
account = Account(space_id=space_id, name="Allowed", kind="cash", opened_on=date.today())
|
|
db.add(account)
|
|
db.flush()
|
|
document = StatementDocument(
|
|
space_id=space_id,
|
|
account_id=account.id,
|
|
original_filename="synthetic.xlsx",
|
|
size_bytes=1,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="confirmed",
|
|
)
|
|
db.add(document)
|
|
db.flush()
|
|
run = ParseRun(
|
|
space_id=space_id,
|
|
document_id=document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(run)
|
|
db.flush()
|
|
document.active_parse_run_id = run.id
|
|
hold = SourceRecord(
|
|
space_id=space_id,
|
|
document_id=document.id,
|
|
parse_run_id=run.id,
|
|
occurred_on=date.today(),
|
|
amount_kopecks=100,
|
|
direction="out",
|
|
bank_status="hold",
|
|
row_ref="sheet0:row1",
|
|
record_fp=uuid.uuid4().hex,
|
|
)
|
|
db.add(hold)
|
|
db.flush()
|
|
db.add(
|
|
HoldResolution(
|
|
space_id=space_id,
|
|
hold_source_record_id=hold.id,
|
|
resolution="discarded_as_cancelled",
|
|
reason="synthetic test resolution",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
set_space_context(db, uuid.uuid4())
|
|
assert db.query(Account).count() == 0
|
|
assert db.query(HoldResolution).count() == 0
|
|
db.rollback()
|
|
set_space_context(db, space_id)
|
|
persisted_space = db.get(FinancialSpace, space_id)
|
|
assert persisted_space is not None
|
|
db.delete(persisted_space)
|
|
db.commit()
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.mark.skipif(not os.getenv("S3_ENDPOINT_URL"), reason="requires S3-compatible storage")
|
|
def test_s3_blob_round_trip() -> None:
|
|
store = get_blob_store()
|
|
key = uuid.uuid4().hex
|
|
payload = b"synthetic statement payload"
|
|
store.put(key, payload)
|
|
assert store.get(key) == payload
|
|
store.delete(key)
|