107 lines
4.8 KiB
Python
107 lines
4.8 KiB
Python
"""Real position-user PUT route against SQLite; never connect to business databases."""
|
|
from datetime import datetime
|
|
|
|
import pymysql
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.api import admin
|
|
from app.models.models import SysPost, SysUser, SysUserPost
|
|
|
|
|
|
@pytest.fixture
|
|
def context(monkeypatch):
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('Business database access is forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
|
|
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
|
for model in (SysPost, SysUser, SysUserPost):
|
|
model.__table__.create(engine)
|
|
db = Session(engine, expire_on_commit=False)
|
|
for uid, tenant, deleted in [('actor', 't', 0), ('u1', 't', 0), ('u2', 't', 0), ('foreign', 'other', 0), ('deleted', 't', 1)]:
|
|
db.add(SysUser(id=uid, user_no=uid, nick_name=uid, tenant_no=tenant, del_flag=deleted))
|
|
for pid, tenant, deleted in [('p', 't', 0), ('p2', 't', 0), ('foreign-p', 'other', 0), ('deleted-p', 't', 1)]:
|
|
db.add(SysPost(id=pid, post_name=pid, post_code=pid, tenant_no=tenant, del_flag=deleted))
|
|
for rid, uid, pid, tenant in [('kept', 'u1', 'p', 't'), ('other-post', 'u1', 'p2', 't'), ('other-tenant', 'foreign', 'foreign-p', 'other')]:
|
|
db.add(SysUserPost(id=rid, user_id=uid, post_id=pid, tenant_no=tenant, del_flag=0,
|
|
create_by='creator', create_time=datetime(2020, 1, 1)))
|
|
db.commit()
|
|
writes, commits = [], []
|
|
def capture(_conn, _cursor, statement, *_args):
|
|
if statement.lstrip().upper().startswith(('INSERT', 'UPDATE', 'DELETE')):
|
|
writes.append(statement)
|
|
event.listen(engine, 'before_cursor_execute', capture)
|
|
event.listen(db, 'after_commit', lambda _db: commits.append(True))
|
|
app = FastAPI()
|
|
app.add_api_route('/posts/{post_id}/users', admin.update_post_bound_users, methods=['PUT'])
|
|
app.dependency_overrides[admin.get_db] = lambda: db
|
|
app.dependency_overrides[admin.require_admin_or_super_admin] = lambda: {'user_id': 'actor'}
|
|
with TestClient(app) as client:
|
|
yield client, db, writes, commits
|
|
db.close()
|
|
engine.dispose()
|
|
|
|
|
|
def put(context, payload, post_id='p'):
|
|
return context[0].put(f'/posts/{post_id}/users', json=payload)
|
|
|
|
|
|
def test_omitted_user_ids_does_not_write_or_commit(context):
|
|
response = put(context, {})
|
|
assert response.status_code == 200
|
|
_, db, writes, commits = context
|
|
assert db.get(SysUserPost, 'kept').del_flag == 0
|
|
assert not writes and not commits
|
|
assert not db.dirty and not db.new
|
|
|
|
|
|
@pytest.mark.parametrize('key', ['userIds', 'user_ids'])
|
|
def test_explicit_empty_array_clears_only_current_post(context, key):
|
|
assert put(context, {key: []}).status_code == 200
|
|
db = context[1]
|
|
assert db.get(SysUserPost, 'kept').del_flag == 1
|
|
assert db.get(SysUserPost, 'other-post').del_flag == 0
|
|
assert db.get(SysUserPost, 'other-tenant').del_flag == 0
|
|
assert db.query(SysUserPost).count() == 3
|
|
|
|
|
|
@pytest.mark.parametrize('key', ['userIds', 'user_ids'])
|
|
def test_explicit_null_is_rejected_before_any_write(context, key):
|
|
assert put(context, {key: None}).status_code == 422
|
|
assert context[1].get(SysUserPost, 'kept').del_flag == 0
|
|
assert not context[2] and not context[3]
|
|
|
|
|
|
def test_nonempty_array_preserves_old_identity_and_adds_only_new_user(context):
|
|
for _ in range(2):
|
|
assert put(context, {'userIds': ['u1', 'u2']}).status_code == 200
|
|
db = context[1]
|
|
kept = db.get(SysUserPost, 'kept')
|
|
assert kept.del_flag == 0 and kept.create_by == 'creator'
|
|
assert kept.create_time == datetime(2020, 1, 1)
|
|
assert db.query(SysUserPost).filter_by(post_id='p', del_flag=0).count() == 2
|
|
assert db.query(SysUserPost).count() == 4
|
|
assert db.get(SysUserPost, 'other-post').del_flag == 0
|
|
assert db.get(SysUserPost, 'other-tenant').del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('uid', ['foreign', 'missing', 'deleted'])
|
|
def test_invalid_user_rejected_before_mutating_existing_bindings(context, uid):
|
|
response = put(context, {'userIds': ['u2', uid]})
|
|
assert response.json()['code'] == 400
|
|
assert context[1].get(SysUserPost, 'kept').del_flag == 0
|
|
assert not context[2] and not context[3]
|
|
|
|
|
|
@pytest.mark.parametrize('post_id', ['foreign-p', 'missing', 'deleted-p'])
|
|
@pytest.mark.parametrize('payload', [{}, {'userIds': []}])
|
|
def test_post_scope_is_validated_even_when_user_ids_are_omitted(context, post_id, payload):
|
|
response = put(context, payload, post_id)
|
|
assert response.json()['code'] == 404
|
|
assert not context[2] and not context[3]
|