Initial commit: InviteMe production redesign

2026-09-01 10:37:40 +03:00
commit d54a17b0ad
71 changed files with 6898 additions and 0 deletions
@@ -0,0 +1,47 @@
+from pathlib import Path
+
+import pytest
+
+from app import create_app, db
+from config import Config
+
+
+@pytest.fixture()
+def app(tmp_path):
+    class TestConfig(Config):
+        TESTING = True
+        IS_PRODUCTION = False
+        SECRET_KEY = "test-secret-key-that-is-long-enough"
+        CREDENTIAL_ENCRYPTION_KEY = "test-encryption-key-that-is-long-enough"
+        SQLALCHEMY_DATABASE_URI = f"sqlite:///{tmp_path / 'test.db'}"
+        SQLALCHEMY_ENGINE_OPTIONS = {}
+        WTF_CSRF_ENABLED = False
+        SESSION_COOKIE_SECURE = False
+        ENFORCE_HTTPS = False
+        ALLOW_WEB_SETUP = True
+        RATELIMIT_ENABLED = False
+        UPLOAD_FOLDER = tmp_path / "uploads"
+        PHOTO_FOLDER = UPLOAD_FOLDER / "photos"
+        QR_FOLDER = UPLOAD_FOLDER / "qrcodes"
+        LOGO_FOLDER = UPLOAD_FOLDER / "logos"
+
+    application = create_app(TestConfig)
+    yield application
+    with application.app_context():
+        db.session.remove()
+        db.drop_all()
+
+
+@pytest.fixture()
+def client(app):
+    return app.test_client()
+
+
+def login_as(client, user, current_event_id=None):
+    with client.session_transaction() as session:
+        session["user_id"] = user.id
+        session["role"] = user.role
+        session["session_version"] = user.session_version
+        if current_event_id:
+            session["current_event_id"] = current_event_id
+            session["checkin_event_id"] = current_event_id
@@ -0,0 +1,300 @@
+import io
+
+import pytest
+from PIL import Image
+from sqlalchemy.exc import IntegrityError
+
+from app import db
+from app.models import AuditLog, CheckIn, Event, Guest, SmtpSettings, User, utcnow
+from app.services.email_service import send_invitation
+from app.tasks import send_event_invitations_in_context
+from app.utils.paths import event_photo_folder
+from config import Config
+
+
+def _user(username, role):
+    user = User(username=username, role=role, is_active=True)
+    user.set_password("Valid-Password-123")
+    db.session.add(user)
+    db.session.flush()
+    return user
+
+
+def _guest(event, email="guest@example.com"):
+    guest = Guest(
+        event_id=event.id,
+        department="IT",
+        full_name="Тестовый Гость",
+        email=email,
+        photo_filename="guest.png",
+    )
+    db.session.add(guest)
+    db.session.flush()
+    return guest
+
+
+def test_organizer_cannot_read_another_event_api(app, client):
+    with app.app_context():
+        owner = _user("owner", "organizer")
+        other = _user("other", "organizer")
+        own_event = Event(name="Own", created_by_id=owner.id)
+        other_event = Event(name="Other", created_by_id=other.id)
+        db.session.add_all([own_event, other_event])
+        db.session.flush()
+        _guest(own_event, "own@example.com")
+        _guest(other_event, "other@example.com")
+        db.session.commit()
+        owner_id = owner.id
+        owner_version = owner.session_version
+        other_event_id = other_event.id
+
+    with client.session_transaction() as session:
+        session["user_id"] = owner_id
+        session["role"] = "organizer"
+        session["session_version"] = owner_version
+
+    response = client.get(f"/api/guests/search?q=Тест&event_id={other_event_id}")
+    assert response.status_code == 403
+
+
+def test_private_photo_requires_event_access(app, client):
+    with app.app_context():
+        owner = _user("owner", "organizer")
+        other = _user("other", "organizer")
+        event = Event(name="Private", created_by_id=owner.id)
+        db.session.add(event)
+        db.session.flush()
+        guest = _guest(event)
+        folder = event_photo_folder(event.id)
+        folder.mkdir(parents=True, exist_ok=True)
+        Image.new("RGB", (10, 10), "white").save(folder / guest.photo_filename)
+        db.session.commit()
+        other_id = other.id
+        other_version = other.session_version
+        guest_id = guest.id
+
+    with client.session_transaction() as session:
+        session["user_id"] = other_id
+        session["role"] = "organizer"
+        session["session_version"] = other_version
+
+    assert client.get(f"/media/guests/{guest_id}/photo").status_code == 403
+
+
+def test_staff_access_requires_event_assignment(app, client):
+    with app.app_context():
+        admin = _user("admin", "admin")
+        staff = _user("staff-assigned", "staff")
+        event = Event(name="Assigned", created_by_id=admin.id)
+        db.session.add(event)
+        db.session.flush()
+        guest = _guest(event)
+        db.session.commit()
+        staff_id = staff.id
+        staff_version = staff.session_version
+        event_id = event.id
+        guest_id = guest.id
+
+    with client.session_transaction() as session:
+        session["user_id"] = staff_id
+        session["role"] = "staff"
+        session["session_version"] = staff_version
+
+    assert client.get(f"/api/guests/search?q=Тест&event_id={event_id}").status_code == 403
+
+    with app.app_context():
+        event = db.session.get(Event, event_id)
+        staff = db.session.get(User, staff_id)
+        event.staff_members.append(staff)
+        db.session.commit()
+
+    response = client.get(f"/api/guests/search?q=Тест&event_id={event_id}")
+    assert response.status_code == 200
+    assert response.get_json()[0]["id"] == guest_id
+
+
+def test_invalid_image_upload_is_not_saved(app, client):
+    with app.app_context():
+        organizer = _user("organizer", "organizer")
+        event = Event(name="Event", created_by_id=organizer.id)
+        db.session.add(event)
+        db.session.commit()
+        organizer_id = organizer.id
+        organizer_version = organizer.session_version
+        event_id = event.id
+
+    with client.session_transaction() as session:
+        session["user_id"] = organizer_id
+        session["role"] = "organizer"
+        session["session_version"] = organizer_version
+        session["current_event_id"] = event_id
+
+    response = client.post(
+        "/admin/upload",
+        data={"photo_files": (io.BytesIO(b"<script>alert(1)</script>"), "photo.jpg")},
+        content_type="multipart/form-data",
+    )
+    assert response.status_code == 302
+    with app.app_context():
+        assert not (event_photo_folder(event_id) / "photo.jpg").exists()
+
+
+def test_mail_password_is_encrypted_at_rest(app):
+    with app.app_context():
+        settings = SmtpSettings.get_settings()
+        settings.set_mail_password("smtp-password")
+        db.session.commit()
+        raw_value = settings.mail_password
+        assert raw_value.startswith("fernet:")
+        assert "smtp-password" not in raw_value
+        assert settings.get_mail_password() == "smtp-password"
+
+
+def test_security_headers_are_present(client):
+    response = client.get("/login")
+    assert response.headers["X-Content-Type-Options"] == "nosniff"
+    assert response.headers["X-Frame-Options"] == "DENY"
+    assert "frame-ancestors 'none'" in response.headers["Content-Security-Policy"]
+
+
+def test_web_setup_can_be_disabled(app, client):
+    app.config["ALLOW_WEB_SETUP"] = False
+    assert client.get("/setup").status_code == 404
+
+
+def test_unsafe_production_config_is_rejected():
+    class UnsafeProductionConfig(Config):
+        IS_PRODUCTION = True
+        SECRET_KEY = "change-me-in-production"
+        CREDENTIAL_ENCRYPTION_KEY = "x" * 32
+        BASE_URL = "https://events.example.com"
+        TRUSTED_HOSTS = ["events.example.com"]
+        SQLALCHEMY_DATABASE_URI = "postgresql+psycopg://example"
+        RATELIMIT_STORAGE_URI = "redis://example"
+        SESSION_COOKIE_SECURE = True
+        ENFORCE_HTTPS = True
+        ALLOW_WEB_SETUP = False
+
+    with pytest.raises(RuntimeError, match="SECRET_KEY"):
+        UnsafeProductionConfig.validate()
+
+
+def test_checkin_api_requires_csrf_when_enabled(app, client):
+    with app.app_context():
+        staff = _user("staff", "staff")
+        event = Event(name="Event", created_by_id=staff.id)
+        db.session.add(event)
+        db.session.flush()
+        guest = _guest(event)
+        db.session.commit()
+        staff_id = staff.id
+        staff_version = staff.session_version
+        token = guest.token
+
+    with client.session_transaction() as session:
+        session["user_id"] = staff_id
+        session["role"] = "staff"
+        session["session_version"] = staff_version
+    app.config["WTF_CSRF_ENABLED"] = True
+
+    assert client.post(f"/api/check-in/{token}").status_code == 400
+
+
+def test_admin_guest_search_and_pagination(app, client):
+    with app.app_context():
+        organizer = _user("search-owner", "organizer")
+        event = Event(name="Search", created_by_id=organizer.id)
+        db.session.add(event); db.session.flush()
+        for number in range(55):
+            guest = _guest(event, f"guest-{number:02d}@example.com")
+            guest.full_name = f"Гость {number:02d}"
+        db.session.commit()
+        user_id, version, event_id = organizer.id, organizer.session_version, event.id
+    with client.session_transaction() as session:
+        session.update(user_id=user_id, role="organizer", session_version=version, current_event_id=event_id)
+    assert client.get("/admin/guests").data.count(b"guest-") == 50
+    assert client.get("/admin/guests?page=2").data.count(b"guest-") == 5
+    assert b"guest-54@example.com" in client.get("/admin/guests?q=guest-54@example.com").data
+    assert client.get("/admin/upload/template").data.startswith(b"PK")
+
+
+def test_duplicate_checkin_is_rejected_and_can_be_undone(app, client):
+    with app.app_context():
+        admin = _user("gate-admin", "admin")
+        staff = _user("gate-staff", "staff")
+        event = Event(name="Gate", created_by_id=admin.id)
+        event.staff_members.append(staff)
+        db.session.add(event); db.session.flush()
+        guest = _guest(event, "gate@example.com")
+        db.session.commit()
+        staff_id, version, token, guest_id = staff.id, staff.session_version, guest.token, guest.id
+    with client.session_transaction() as session:
+        session.update(user_id=staff_id, role="staff", session_version=version)
+    assert client.post(f"/api/check-in/{token}").status_code == 200
+    assert client.post(f"/api/check-in/{token}").status_code == 409
+    assert client.post(f"/api/check-in/{token}/undo", json={"reason": "Ошибка оператора"}).status_code == 200
+    with app.app_context():
+        assert CheckIn.query.filter_by(guest_id=guest_id).count() == 0
+        assert AuditLog.query.filter_by(action="check_in_undo", target_id=str(guest_id)).count() == 1
+
+
+def test_database_rejects_duplicate_checkins(app):
+    with app.app_context():
+        owner = _user("constraint-owner", "admin")
+        event = Event(name="Constraint", created_by_id=owner.id)
+        db.session.add(event); db.session.flush()
+        guest = _guest(event, "constraint@example.com")
+        db.session.flush()
+        db.session.add_all([CheckIn(guest_id=guest.id, event_id=event.id), CheckIn(guest_id=guest.id, event_id=event.id)])
+        with pytest.raises(IntegrityError):
+            db.session.commit()
+        db.session.rollback()
+
+
+def test_public_invitation_rsvp_calendar_and_ticket(app, client, monkeypatch, tmp_path):
+    with app.app_context():
+        owner = _user("rsvp-owner", "organizer")
+        event = Event(
+            name="RSVP Event", created_by_id=owner.id, event_date=utcnow(),
+            event_end_date=utcnow(), rsvp_enabled=True,
+        )
+        db.session.add(event); db.session.flush()
+        guest = _guest(event, "rsvp@example.com")
+        db.session.commit()
+        token, guest_id = guest.token, guest.id
+
+    assert client.get(f"/invite/{token}").status_code == 200
+    response = client.post(f"/invite/{token}", data={
+        "rsvp_status": "confirmed",
+        "dietary_requirements": "Без орехов",
+        "rsvp_comment": "Буду вовремя",
+    })
+    assert response.status_code == 302
+    calendar = client.get(f"/invite/{token}/calendar.ics")
+    assert calendar.status_code == 200
+    assert b"BEGIN:VCALENDAR" in calendar.data
+
+    ticket_path = tmp_path / "ticket.png"
+    Image.new("RGB", (10, 10), "white").save(ticket_path)
+    monkeypatch.setattr("app.routes.generate_ticket_image", lambda *_args, **_kwargs: ticket_path)
+    assert client.get(f"/invite/{token}/ticket").status_code == 200
+    with app.app_context():
+        updated = db.session.get(Guest, guest_id)
+        assert updated.rsvp_status == "confirmed"
+        assert updated.dietary_requirements == "Без орехов"
+
+
+def test_invitation_delivery_and_worker_status(app, monkeypatch):
+    with app.app_context():
+        owner = _user("worker-owner", "organizer")
+        event = Event(name="Worker", created_by_id=owner.id)
+        db.session.add(event); db.session.flush()
+        guest = _guest(event, "worker@example.com")
+        db.session.commit()
+        monkeypatch.setattr("app.services.email_service.generate_ticket_image", lambda *_args, **_kwargs: None)
+        monkeypatch.setattr("app.services.email_service._send_email", lambda **_kwargs: None)
+        result = send_event_invitations_in_context(event.id, owner.id)
+        db.session.refresh(guest)
+        assert result["sent"] == 1
+        assert guest.invitation_status == "sent"
+        assert guest.invitation_attempts == 1