# Path: amivapi/cron.py
# def run_scheduled_tasks():
#     """ Check for scheduled task, which have passed the deadline and run them.
#     This needs an app context.
#     """
#     while True:
#         task = (current_app.data.driver.db['scheduled_tasks']
#                 .find_one_and_delete(
#                     {'time': {'$lte': datetime.utcnow()}}))
# 
#         if task is None:
#             return
# 
#         args = pickle.loads(task['args'])
#         func = schedulable_functions[task['function']]
#         func(*args)
# 
# Path: amivapi/tests/utils.py
# class WebTest(unittest.TestCase, FixtureMixin):
#     """Base test class for tests against the full WSGI stack.
# 
#     Inspired by eve standard testing class.
#     """
# 
#     # Test Config overwrites
#     test_config = {
#         'MONGO_DBNAME': 'test_amivapi',
#         'MONGO_USERNAME': 'test_user',
#         'MONGO_PASSWORD': 'test_pw',
#         'API_MAIL': 'api@test.ch',
#         'SMTP_SERVER': '',
#         'TESTING': True,
#         'DEBUG': True,   # This makes eve's error messages more helpful
#         'LDAP_USERNAME': None,  # LDAP test require special treatment
#         'LDAP_PASSWORD': None,  # LDAP test require special treatment
#         'SENTRY_DSN': None,
#         'SENTRY_ENVIRONMENT': None,
#         'PASSWORD_CONTEXT': CryptContext(
#             schemes=["pbkdf2_sha256"],
#             pbkdf2_sha256__default_rounds=10,
#             # min_rounds is used to determine if a hash needs to be upgraded
#             pbkdf2_sha256__min_rounds=8,
#         )
#     }
# 
#     def setUp(self, **extra_config):
#         """Set up the testing client and database connection.
# 
#         self.api will be a flask TestClient to make requests
#         self.db will be a MongoDB database
#         """
#         super().setUp()
# 
#         # In 3.2, assertItemsEqual was replaced by assertCountEqual
#         # Make assertItemsEqual work in tests for py3 as well
#         if sys.version_info >= (3, 2):
#             self.assertItemsEqual = self.assertCountEqual
# 
#         # create eve app and test client
#         config = {}
#         config.update(self.test_config)
#         config.update(extra_config)
#         self.app = bootstrap.create_app(**config)
#         self.app.response_class = TestResponse
#         self.app.test_client_class = TestClient
#         self.app.test_mails = []
#         self.api = self.app.test_client()
# 
#         # Create a separate mongo connection and db reference for tests
#         self.connection = MongoClient(host=self.app.config['MONGO_HOST'],
#                                       port=self.app.config['MONGO_PORT'])
#         self.db = self.connection[self.app.config['MONGO_DBNAME']]
#         self.db.authenticate(name=self.app.config['MONGO_USERNAME'],
#                              password=self.app.config['MONGO_PASSWORD'],
#                              source=self.app.config['MONGO_DBNAME'])
# 
#     def tearDown(self):
#         """Tear down after testing."""
#         # delete testing database
#         self.connection.drop_database(self.test_config['MONGO_DBNAME'])
#         # close database connection
#         self.connection.close()
# 
#     # Shortcuts to get a token
#     counter = count()
# 
#     def get_user_token(self, user_id, created=None):
#         """Create session for a user and return a token.
# 
#         Args:
#             user_id (str): user_id as string.
# 
#         Returns:
#             str: Token that can be used to authenticate user.
#         """
#         if created is None:
#             created = datetime.now(timezone.utc)
# 
#         token = "test_token_" + str(next(self.counter))
#         self.db['sessions'].insert_one({u'user': ObjectId(user_id),
#                                         u'token': token,
#                                         u'_created': created})
#         return token
# 
#     def get_root_token(self):
#         """The root password is the root token.
# 
#         Returns:
#             str: Token for the root user
#         """
#         return ROOT_PASSWORD


from datetime import timedelta
from freezegun import freeze_time
from amivapi.cron import run_scheduled_tasks
from amivapi.tests.utils import WebTest
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# license: AGPLv3, see LICENSE for details. In addition we strongly encourage
#          you to buy us beer if we meet and you like the software.

""" Test that sessions get cleaned up after enough time passed. """




class TestSessionExpiry(WebTest):
    def test_session_expiry(self):
        with self.app.app_context(), freeze_time() as frozen_time:
            self.new_object("users", nethz="pablo", password="password")
            self.api.post('/sessions',
                          data={"username": "pablo", "password": "password"},
                          status_code=201)

            frozen_time.tick(delta=self.app.config['SESSION_TIMEOUT'] -
                             timedelta(days=1))
            run_scheduled_tasks()

            self.assertEqual(
                self.app.data.driver.db['sessions'].count_documents({}), 1)

            frozen_time.tick(delta=timedelta(days=2))

# Path: amivapi/cron.py
# def run_scheduled_tasks():
#     """ Check for scheduled task, which have passed the deadline and run them.
#     This needs an app context.
#     """
#     while True:
#         task = (current_app.data.driver.db['scheduled_tasks']
#                 .find_one_and_delete(
#                     {'time': {'$lte': datetime.utcnow()}}))
# 
#         if task is None:
#             return
# 
#         args = pickle.loads(task['args'])
#         func = schedulable_functions[task['function']]
#         func(*args)
# 
# Path: amivapi/tests/utils.py
# class WebTest(unittest.TestCase, FixtureMixin):
#     """Base test class for tests against the full WSGI stack.
# 
#     Inspired by eve standard testing class.
#     """
# 
#     # Test Config overwrites
#     test_config = {
#         'MONGO_DBNAME': 'test_amivapi',
#         'MONGO_USERNAME': 'test_user',
#         'MONGO_PASSWORD': 'test_pw',
#         'API_MAIL': 'api@test.ch',
#         'SMTP_SERVER': '',
#         'TESTING': True,
#         'DEBUG': True,   # This makes eve's error messages more helpful
#         'LDAP_USERNAME': None,  # LDAP test require special treatment
#         'LDAP_PASSWORD': None,  # LDAP test require special treatment
#         'SENTRY_DSN': None,
#         'SENTRY_ENVIRONMENT': None,
#         'PASSWORD_CONTEXT': CryptContext(
#             schemes=["pbkdf2_sha256"],
#             pbkdf2_sha256__default_rounds=10,
#             # min_rounds is used to determine if a hash needs to be upgraded
#             pbkdf2_sha256__min_rounds=8,
#         )
#     }
# 
#     def setUp(self, **extra_config):
#         """Set up the testing client and database connection.
# 
#         self.api will be a flask TestClient to make requests
#         self.db will be a MongoDB database
#         """
#         super().setUp()
# 
#         # In 3.2, assertItemsEqual was replaced by assertCountEqual
#         # Make assertItemsEqual work in tests for py3 as well
#         if sys.version_info >= (3, 2):
#             self.assertItemsEqual = self.assertCountEqual
# 
#         # create eve app and test client
#         config = {}
#         config.update(self.test_config)
#         config.update(extra_config)
#         self.app = bootstrap.create_app(**config)
#         self.app.response_class = TestResponse
#         self.app.test_client_class = TestClient
#         self.app.test_mails = []
#         self.api = self.app.test_client()
# 
#         # Create a separate mongo connection and db reference for tests
#         self.connection = MongoClient(host=self.app.config['MONGO_HOST'],
#                                       port=self.app.config['MONGO_PORT'])
#         self.db = self.connection[self.app.config['MONGO_DBNAME']]
#         self.db.authenticate(name=self.app.config['MONGO_USERNAME'],
#                              password=self.app.config['MONGO_PASSWORD'],
#                              source=self.app.config['MONGO_DBNAME'])
# 
#     def tearDown(self):
#         """Tear down after testing."""
#         # delete testing database
#         self.connection.drop_database(self.test_config['MONGO_DBNAME'])
#         # close database connection
#         self.connection.close()
# 
#     # Shortcuts to get a token
#     counter = count()
# 
#     def get_user_token(self, user_id, created=None):
#         """Create session for a user and return a token.
# 
#         Args:
#             user_id (str): user_id as string.
# 
#         Returns:
#             str: Token that can be used to authenticate user.
#         """
#         if created is None:
#             created = datetime.now(timezone.utc)
# 
#         token = "test_token_" + str(next(self.counter))
#         self.db['sessions'].insert_one({u'user': ObjectId(user_id),
#                                         u'token': token,
#                                         u'_created': created})
#         return token
# 
#     def get_root_token(self):
#         """The root password is the root token.
# 
#         Returns:
#             str: Token for the root user
#         """
#         return ROOT_PASSWORD


# Path: amivapi/tests/auth/test_session_expiry.py
from datetime import timedelta
from freezegun import freeze_time
from amivapi.cron import run_scheduled_tasks
from amivapi.tests.utils import WebTest

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# license: AGPLv3, see LICENSE for details. In addition we strongly encourage
#          you to buy us beer if we meet and you like the software.

""" Test that sessions get cleaned up after enough time passed. """




class TestSessionExpiry(WebTest):
    def test_session_expiry(self):
        with self.app.app_context(), freeze_time() as frozen_time:
            self.new_object("users", nethz="pablo", password="password")
            self.api.post('/sessions',
                          data={"username": "pablo", "password": "password"},
                          status_code=201)

            frozen_time.tick(delta=self.app.config['SESSION_TIMEOUT'] -
                             timedelta(days=1))
            run_scheduled_tasks()

            self.assertEqual(
                self.app.data.driver.db['sessions'].count_documents({}), 1)

            frozen_time.tick(delta=timedelta(days=2))

            run_scheduled_tasks()