diff --git a/ambuda/models/auth.py b/ambuda/models/auth.py index c4d7692b..867bd823 100644 --- a/ambuda/models/auth.py +++ b/ambuda/models/auth.py @@ -81,6 +81,40 @@ def __repr__(self): return f"" +class UserStatusLog(AmbudaUserMixin, Base): + """Tracks changes to user statuses.""" + + __tablename__ = "user_status_log" + + #: Primary key. + id = pk() + + #: The user whose status was changed. + user_id = Column(Integer, ForeignKey("users.id"), nullable=False) + + #: Timestamp at which this status change occured. + timestamp = Column(DateTime, default=datetime.utcnow, nullable=False) + + #: Describes the status change that occurred. + change_description = Column(String, nullable=False) + + #: When should this status change expire/revert, defaults to never. + expiry = Column(DateTime, default=None, nullable=True) + + @property + def is_expired(self) -> bool: + """Check if the action has expired.""" + return self.expiry and self.expiry < datetime.utcnow() + + @property + def is_temporary(self) -> bool: + """ + Check if the action has an expiry and will be reverted + in the future. + """ + return self.expiry + + class UserRoles(Base): """Secondary table for users and roles.""" diff --git a/migrations/versions/e4c982111325_create_user_status_log_table.py b/migrations/versions/e4c982111325_create_user_status_log_table.py new file mode 100644 index 00000000..54f81656 --- /dev/null +++ b/migrations/versions/e4c982111325_create_user_status_log_table.py @@ -0,0 +1,40 @@ +"""Create user status log table + +Revision ID: e4c982111325 +Revises: 99379162619e +Create Date: 2022-09-24 19:04:16.240264 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy import orm + + +# revision identifiers, used by Alembic. +revision = 'e4c982111325' +down_revision = '99379162619e' +branch_labels = None +depends_on = None + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "user_status_log", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("timestamp", sa.DateTime(), nullable=False), + sa.Column("change_description", sa.String(), nullable=False), + sa.Column("expiry", sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint( + ["user_id"], + ["users.id"], + ), + sa.PrimaryKeyConstraint("id"), + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("user_status_log") + # ### end Alembic commands ### \ No newline at end of file diff --git a/test/ambuda/test_models.py b/test/ambuda/test_models.py index 4b8651bb..0eff9345 100644 --- a/test/ambuda/test_models.py +++ b/test/ambuda/test_models.py @@ -1,3 +1,4 @@ +from datetime import datetime import ambuda.database as db from ambuda.queries import get_session @@ -85,3 +86,54 @@ def test_token__set_and_check_token(client): assert not row.check_token("password2") _cleanup(session, row) + + +def test_user_status_log__permanent(client): + session = get_session() + row = db.UserStatusLog( + user_id=1, + change_description="action_one", + ) + + session.add(row) + session.commit() + + assert not row.is_expired + assert not row.is_temporary + + _cleanup(session, row) + + +def test_user_status_log__temporary_unexpired(client): + session = get_session() + row = db.UserStatusLog( + user_id=1, + change_description="action_one", + expiry=datetime.fromisoformat('2055-09-22') + ) + + session.add(row) + session.commit() + + assert not row.is_expired + assert row.is_temporary + + _cleanup(session, row) + + +def test_user_status_log__temporary_expired(client): + session = get_session() + row = db.UserStatusLog( + user_id=1, + change_description="action_one", + expiry=datetime.fromisoformat('2020-09-22') + ) + + session.add(row) + session.commit() + + assert row.is_expired + assert row.is_temporary + + _cleanup(session, row) +