CLI - add command to clean expired oauth2 tokens

fittrackee/cli/__init__.py

@@ -1,6 +1,7 @@
 import click
 
 from fittrackee.migrations.commands import db_cli
+from fittrackee.oauth2.commands import oauth2_cli
 from fittrackee.users.commands import users_cli
 
 
@@ -11,4 +12,5 @@ def cli() -> None:
 
 
 cli.add_command(db_cli)
+cli.add_command(oauth2_cli)
 cli.add_command(users_cli)

fittrackee/oauth2/clean.py

@@ -0,0 +1,13 @@
+import time
+
+from fittrackee import db
+
+
+def clean_tokens(days: int) -> int:
+    limit = int(time.time()) - (days * 86400)
+    sql = """
+        DELETE FROM oauth2_token
+        WHERE oauth2_token.issued_at + oauth2_token.expires_in < %(limit)s;
+    """
+    result = db.engine.execute(sql, {'limit': limit})
+    return result.rowcount

fittrackee/oauth2/commands.py

@@ -0,0 +1,29 @@
+import logging
+
+import click
+
+from fittrackee.cli.app import app
+
+from .clean import clean_tokens
+
+handler = logging.StreamHandler()
+logger = logging.getLogger('fittrackee_clean_tokens')
+logger.setLevel(logging.INFO)
+logger.addHandler(handler)
+
+
+@click.group(name='oauth2')
+def oauth2_cli() -> None:
+    """Manage OAuth2 tokens."""
+    pass
+
+
+@oauth2_cli.command('clean')
+@click.option('--days', type=int)
+def clean(
+    days: int,
+) -> None:
+    """Clean tokens expired for more than provided number of days"""
+    with app.app_context():
+        deleted_rows = clean_tokens(days)
+        logger.info(f'Expired deleted tokens: {deleted_rows}.')

fittrackee/tests/oauth2/test_oauth2_clean.py

@@ -0,0 +1,59 @@
+import time
+
+from flask import Flask
+
+from fittrackee.oauth2.clean import clean_tokens
+from fittrackee.oauth2.models import OAuth2Token
+from fittrackee.users.models import User
+
+from ..mixins import OAuth2Mixin
+
+
+class TestOAuth2CleanTokens(OAuth2Mixin):
+    def test_it_does_not_delete_not_expired_token(
+        self, app: Flask, user_1: User
+    ) -> None:
+        oauth_client = self.create_oauth_client(user_1)
+        self.create_oauth2_token(oauth_client)
+
+        clean_tokens(days=1)
+
+        assert OAuth2Token.query.count() == 1
+
+    def test_it_deletes_expired_token(self, app: Flask, user_1: User) -> None:
+        oauth_client = self.create_oauth_client(user_1)
+        expires_in = 864000  # 10 days
+        days = 5
+        self.create_oauth2_token(
+            oauth_client,
+            issued_at=int(time.time()) - expires_in - (days * 86400) - 1,
+            expires_in=expires_in,
+        )
+
+        clean_tokens(days=days)
+
+        assert OAuth2Token.query.count() == 0
+
+    def test_it_returns_deleted_rows_count(
+        self, app: Flask, user_1: User
+    ) -> None:
+        oauth_client = self.create_oauth_client(user_1)
+        expires_in = 86400  # 10 days
+        days = 5
+        expected_deleted_rows = 3
+        for _ in range(expected_deleted_rows):
+            self.create_oauth2_token(
+                oauth_client,
+                issued_at=(int(time.time()) - expires_in - (days * 86400) - 1),
+                expires_in=expires_in,
+            )
+        self.create_oauth2_token(oauth_client)
+        self.create_oauth2_token(
+            oauth_client,
+            issued_at=(int(time.time()) - expires_in - (days * 86400)),
+            expires_in=expires_in,
+        )
+
+        result = clean_tokens(days=days)
+
+        assert result == expected_deleted_rows