added unitests for utcnow_timestamp

Author: voith

frontera/contrib/backends/cassandra/components.py

@@ -132,8 +132,8 @@ def get_next_requests(self, max_n_requests, partition_id, **kwargs):
         """
         results = []
         try:
-            for item in self._order_by(self.queue_model.filter(partition_id=partition_id).
-                                               allow_filtering()).limit(max_n_requests):
+            for item in self._order_by(self.queue_model.filter(partition_id=partition_id).allow_filtering()).\
+                    limit(max_n_requests):
                 method = item.method or b'GET'
                 r = Request(item.url, method=method, meta=item.meta, headers=item.headers, cookies=item.cookies)
                 r.meta[b'fingerprint'] = to_bytes(item.fingerprint)
@@ -208,8 +208,8 @@ def get_next_requests(self, max_n_requests, partition_id, **kwargs):
                               tries, limit, count, len(queue.keys()))
             queue.clear()
             count = 0
-            for item in self._order_by(self.queue_model.filter(partition_id=partition_id).
-                                               allow_filtering()).limit(max_n_requests):
+            for item in self._order_by(self.queue_model.filter(partition_id=partition_id).allow_filtering()).\
+                    limit(max_n_requests):
                 if item.host_crc32 not in queue:
                     queue[item.host_crc32] = []
                 if max_requests_per_host is not None and len(queue[item.host_crc32]) > max_requests_per_host:

tests/contrib/backends/cassandra/test_backend_cassandra.py

@@ -4,7 +4,6 @@
 from time import time
 
 import six
-from cassandra.cluster import Cluster
 from cassandra.cqlengine import connection
 from cassandra.cqlengine.management import (create_keyspace_simple,
                                             drop_keyspace, drop_table,
@@ -40,19 +39,13 @@ def setUp(self):
         self.manager.settings = settings
         self.keyspace = settings.CASSANDRABACKEND_KEYSPACE
         self.timeout = settings.CASSANDRABACKEND_REQUEST_TIMEOUT
-        cluster = Cluster(self.hosts, self.port)
-        self.session = cluster.connect()
         self._set_global_connection(self.hosts, self.port, self.timeout)
         create_keyspace_simple(self.keyspace, 1)
-        self.session.set_keyspace(self.keyspace)
-        self.session.default_timeout = self.timeout
         connection.session.set_keyspace(self.keyspace)
 
     def tearDown(self):
         self._set_global_connection(self.hosts, self.port, self.timeout)
         drop_keyspace(self.keyspace)
-        self.session.shutdown()
-        connection.unregister_connection('default')
 
     def _set_global_connection(self, hosts, port, timeout):
         if not connection.cluster:
@@ -141,9 +134,9 @@ def assert_db_values(self, model, _filter, fields):
 class TestCassandraBackend(BaseCassandraTest, unittest.TestCase):
 
     def _get_tables(self):
-        query = self.session.prepare('SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?')
-        result = self.session.execute(query, (self.session.keyspace,))
-        return [row.table_name for row in result.current_rows]
+        query = 'SELECT table_name FROM system_schema.tables WHERE keyspace_name = \'{}\''.format(self.keyspace)
+        result = connection.execute(query)
+        return [row['table_name'] for row in result.current_rows]
 
     def test_tables_created(self):
         tables_before = self._get_tables()

tests/test_utils_misc.py

@@ -1,8 +1,11 @@
 from __future__ import absolute_import
 import hashlib
 import pytest
-from frontera.utils.misc import load_object, get_crc32, chunks, to_signed32
 import six
+from datetime import datetime
+
+from frontera.utils.misc import load_object, get_crc32, chunks, to_signed32, utcnow_timestamp
+from tests import mock
 
 
 class TestGetCRC32(object):
@@ -82,3 +85,13 @@ def test_name_error(self):
             load_object('tests.mocks.load_objects.non_existent_object')
         assert str(info.value) == ("Module 'tests.mocks.load_objects' doesn't define"
                                    " any object named 'non_existent_object'")
+
+
+class TestUtcNowTimestamp(object):
+
+    def test(self):
+        udt = datetime(2016, 11, 11, 0, 0, 0)
+        with mock.patch('frontera.utils.misc.datetime') as mocked_datetime:
+            mocked_datetime.utcnow = mock.Mock(return_value=udt)
+            utc_tstamp = utcnow_timestamp()
+            assert utc_tstamp == 1478822400