Merge pull request #70 from aaron-suarez/timestamps

Add create and update timestamps to Resource objects

Author: apex-omontgomery

.env

@@ -1,3 +1,4 @@
 FLASK_APP=run.py
-SQLALCHEMY_DATABASE_URI=postgresql://user_name:change_password@127.0.0.1:5432/resources
+SQLALCHEMY_DATABASE_URI=postgresql://aaron:getmein@127.0.0.1:5432/resources
 FLASK_SKIP_DOTENV=1
+FLASK_ENV=development

app/api/routes.py

@@ -1,13 +1,18 @@
 from traceback import print_tb
 
 from flask import request
-from sqlalchemy import and_, func
+from sqlalchemy import or_, func
 from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
 
 from app.api import bp
 from app.models import Language, Resource, Category
 from app import Config, db
 from app.utils import Paginator, standardize_response
+from dateutil import parser
+from datetime import datetime
+import logging
+
+logger = logging.getLogger()
 
 
 # Routes
@@ -72,43 +77,53 @@ def get_resources():
     # Fetch the filter params from the url, if they were provided.
     language = request.args.get('language')
     category = request.args.get('category')
+    updated_after = request.args.get('updated_after')
+
+    q = Resource.query
 
     # Filter on language
-    if language and not category:
-        query = Resource.query.filter(
+    if language:
+        q = q.filter(
             Resource.languages.any(
                 Language.name.ilike(language)
             )
         )
 
     # Filter on category
-    elif category and not language:
-        query = Resource.query.filter(
+    if category:
+        q = q.filter(
             Resource.category.has(
                 func.lower(Category.name) == category.lower()
             )
         )
 
-    # Filter on both
-    elif category and language:
-        query = Resource.query.filter(
-            and_(
-                Resource.languages.any(
-                    Language.name.ilike(language)
-                ),
-                Resource.category.has(
-                    func.lower(Category.name) == category.lower()
-                )
+    # Filter on updated_after
+    if updated_after:
+        try:
+            uaDate = parser.parse(updated_after)
+            if uaDate > datetime.now():
+                raise Exception("updated_after greater than today's date")
+            uaDate = uaDate.strftime("%Y-%m-%d")
+        except Exception as e:
+            logger.error(e)
+            message = 'The value for "updated_after" is invalid'
+            error = [{"code": "bad-value", "message": message}]
+            return standardize_response(None, error, "unprocessable-entity", 422)
+
+        q = q.filter(
+            or_(
+                Resource.created_at >= uaDate,
+                Resource.last_updated >= uaDate
             )
         )
 
-    # No filters
-    else:
-        query = Resource.query
-
-    resource_list = [
-        resource.serialize for resource in resource_paginator.items(query)
-    ]
+    try:
+        resource_list = [
+            resource.serialize for resource in resource_paginator.items(q)
+        ]
+    except Exception as e:
+        logger.error(e)
+        return standardize_response(None, [{"code": "bad-request"}], "bad request", 400)
 
     return standardize_response(resource_list, None, "ok")
 

app/models.py

@@ -1,5 +1,7 @@
 from app import db
 from sqlalchemy_utils import URLType
+from sqlalchemy import DateTime
+from sqlalchemy.sql import func
 
 
 language_identifier = db.Table('language_identifier',
@@ -28,6 +30,8 @@ class Resource(db.Model):
     upvotes = db.Column(db.INTEGER, default=0)
     downvotes = db.Column(db.INTEGER, default=0)
     times_clicked = db.Column(db.INTEGER, default=0)
+    created_at = db.Column(DateTime(timezone=True), server_default=func.now())
+    last_updated = db.Column(DateTime(timezone=True), onupdate=func.now())
 
     @property
     def serialize(self):

app/utils.py

@@ -16,15 +16,15 @@ def items(self, query):
         return query.paginate(self.page, self.page_size, False).items
 
 
-def standardize_response(data, errors, status):
+def standardize_response(data, errors, status, status_code=200):
     resp = {
         "status": status,
         "apiVersion": API_VERSION
     }
-    if data:
+    if data is not None:
         resp["data"] = data
     elif errors:
         resp["errors"] = errors
     else:
         resp["errors"] = [{"code": "something-went-wrong"}]
-    return jsonify(resp)
+    return jsonify(resp), status_code

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts = -p no:warnings

tests/conftest.py

@@ -1,7 +1,58 @@
 import pytest
-from app import create_app
+from app import create_app, db as _db
+from configs import Config
 
-@pytest.fixture(scope='module')
-def app():
-    app = create_app()
-    return app
+TEST_DATABASE_URI = 'sqlite:///:memory:'
+
+counter = 0
+
+
+@pytest.fixture(scope='session')
+def app(request):
+    Config.SQLALCHEMY_DATABASE_URI = TEST_DATABASE_URI
+    Config.TESTING = True
+    app = create_app(Config)
+
+
+    # Establish an application context before running the tests.
+    ctx = app.app_context()
+    ctx.push()
+
+    def teardown():
+        ctx.pop()
+
+    request.addfinalizer(teardown)
+    return app
+
+
+@pytest.fixture(scope='session')
+def db(app, request):
+    """Session-wide test database."""
+    def teardown():
+        _db.drop_all()
+
+    _db.app = app
+    _db.create_all()
+
+    request.addfinalizer(teardown)
+    return _db
+
+
+@pytest.fixture(scope='function')
+def session(db, request):
+    """Creates a new database session for a test."""
+    connection = db.engine.connect()
+    transaction = connection.begin()
+
+    options = dict(bind=connection, binds={})
+    session = db.create_scoped_session(options=options)
+
+    db.session = session
+
+    def teardown():
+        transaction.rollback()
+        connection.close()
+        session.remove()
+
+    request.addfinalizer(teardown)
+    return session

tests/unit/test_routes.py

@@ -1,11 +1,171 @@
 import pytest
+from tests import conftest
 from app.models import Resource, Language, Category
+from configs import PaginatorConfig
+from app.cli import import_resources
 
 
-def test_does_nothing():
-    """
-    GIVEN a User model
-    WHEN a new User is created
-    THEN check the email, hashed_password, authenticated, and role fields are defined correctly
-    """
-    assert(1 == 1)
+##########################################
+## Tests
+##########################################
+
+# TODO: We need negative unit tests (what happens when bad data is sent)
+
+def test_getters(app, session, db):
+    # Importing the data takes a rather long time, so import it once
+    # here for all the getters that require pages of resources
+    import_resources(db)
+    client = app.test_client()
+
+    # Actually conduct all the tests in helper functions
+    get_resources_test(app, session, db, client)
+    paginator_test(app, session, db, client)
+    filters_test(app, session, db, client)
+    languages_test(app, session, db, client)
+    categories_test(app, session, db, client)
+
+
+def test_create_resource(app, session, db):
+    client = app.test_client()
+
+    response = create_resource(client)
+    assert(response.status_code == 200)
+    assert(isinstance(response.json['data'].get('id'), int))
+    assert(response.json['data'].get('name') == "Some Name")
+
+
+def test_update_resource(app, session, db):
+    client = app.test_client()
+
+    response = create_resource(client)
+
+    id = response.json['data'].get('id')
+    assert(isinstance(id, int))
+
+    response = client.put(f"/api/v1/resources/{id}", json={
+        "name": "New name"
+    })
+
+    assert(response.status_code == 200)
+    assert(response.json['data'].get('name') == "New name")
+
+
+##########################################
+## Helpers
+##########################################
+
+def get_resources_test(app, session, db, client):
+    response = client.get('api/v1/resources')
+
+    # Status should be OK
+    assert(response.status_code == 200)
+
+    resources = response.json
+
+    # Default page size shouold be specified in PaginatorConfig
+    assert(len(resources['data']) == PaginatorConfig.per_page)
+    check_resources(resources['data'])
+
+
+def get_single_resource_test(app, session, db, client):
+    response = client.get('api/v1/resources/5')
+
+    # Status should be OK
+    assert(response.status_code == 200)
+
+    resource = response.json
+
+    check_resources([resources['data']])
+    assert(resources['data'].get('id') == 5)
+
+
+
+def paginator_test(app, session, db, client):
+    # Test page size
+    response = client.get('api/v1/resources?page_size=1')
+    assert(len(response.json['data']) == 1)
+    response = client.get('api/v1/resources?page_size=5')
+    assert(len(response.json['data']) == 5)
+    response = client.get('api/v1/resources?page_size=10')
+    assert(len(response.json['data']) == 10)
+    response = client.get('api/v1/resources?page_size=100')
+    assert(len(response.json['data']) == 100)
+
+    # Test pages different and sequential
+    first_page_resource = response.json['data'][0]
+    assert(first_page_resource.get('id') == 1)
+    response = client.get('api/v1/resources?page_size=100&page=2')
+    second_page_resource = response.json['data'][0]
+    assert(second_page_resource.get('id') == 101)
+    response = client.get('api/v1/resources?page_size=100&page=3')
+    third_page_resource = response.json['data'][0]
+    assert(third_page_resource.get('id') == 201)
+
+    # Test bigger than max page size
+    too_long = PaginatorConfig.max_page_size + 1
+    response = client.get(f"api/v1/resources?page_size={too_long}")
+    assert(len(response.json['data']) == PaginatorConfig.max_page_size)
+
+    # Test farther than last page
+    too_far = 99999999
+    response = client.get(f"api/v1/resources?page_size=100&page={too_far}")
+    assert(len(response.json['data']) == 0)
+
+
+def filters_test(app, session, db, client):
+    # Filter by language
+    response = client.get('api/v1/resources?language=python')
+
+    for resource in response.json['data']:
+        assert(type(resource.get('languages')) is list)
+        assert('Python' in resource.get('languages'))
+
+    # Filter by category
+    response = client.get('api/v1/resources?category=Back%20End%20Dev')
+
+    for resource in response.json['data']:
+        assert(resource.get('category') == "Back End Dev")
+
+    # TODO: Filter by updated_after
+    # (Need to figure out how to manually set last_updated and created_at)
+
+
+def languages_test(app, session, db, client):
+    response = client.get('api/v1/languages')
+
+    for language in response.json['data']:
+        assert(isinstance(language.get('id'), int))
+        assert(isinstance(language.get('name'), str))
+        assert(len(language.get('name')) > 0)
+
+
+def categories_test(app, session, db, client):
+    response = client.get('api/v1/categories')
+
+    for category in response.json['data']:
+        assert(isinstance(category.get('id'), int))
+        assert(isinstance(category.get('name'), str))
+        assert(len(category.get('name')) > 0)
+
+
+def check_resources(resources):
+    # Each resource should have a name, url, languages and category
+    for resource in resources:
+        assert(isinstance(resource.get('name'), str))
+        assert(resource.get('name') != "")
+        assert(isinstance(resource.get('url'), str))
+        assert(resource.get('url') != "")
+        assert(isinstance(resource.get('category'), str))
+        assert(resource.get('category') != "")
+        assert(type(resource.get('languages')) is list)
+
+
+def create_resource(client):
+    return client.post('/api/v1/resources', json={
+        "name": "Some Name",
+        "url": "http://example.org/",
+        "category": "New Category",
+        "languages": ["Python", "New Language"],
+        "paid": False,
+        "notes": "Some notes"
+    })