Merge pull request #318 from algolia/test/pytest

tests: change nose2 for pytest

Author: LeoErcolanelli

.gitignore

@@ -27,3 +27,7 @@ pip-log.txt
 # IDE/OS Files
 .idea
 .DS_Store
+.vscode
+pytest.ini
+tags
+.cache

requirements.txt

@@ -2,7 +2,6 @@ tox
 wheel
 pypandoc
 
-# Dev
 requests[security]>=2.9.1
-nose2
 faker
+pytest

tests/__init__.py

@@ -0,0 +1,50 @@
+import pytest
+
+from helpers import Factory
+from helpers import create_client, create_index, IndexWithData
+
+
+@pytest.fixture
+def mcm_client():
+    return create_client('ALGOLIA_APPLICATION_ID_MCM', 'ALGOLIA_API_KEY_MCM')
+
+
+@pytest.fixture
+def client():
+    return create_client()
+
+
+@pytest.fixture
+def index(client):
+    idx = create_index(client)
+    yield idx
+    client.delete_index(idx.index_name)  # Tear down
+
+
+@pytest.fixture(scope='module')
+def ro_index():
+    idx = IndexWithData(create_client())
+    yield idx
+    idx.client.delete_index(idx.index_name)  # Tear down
+
+
+@pytest.fixture
+def rw_index():
+    idx = IndexWithData(create_client())
+    yield idx
+    idx.client.delete_index(idx.index_name)  # Tear down
+
+
+@pytest.fixture
+def double_indexes():
+    clt = create_client()
+    factory = Factory()
+
+    index1 = IndexWithData(clt, factory=factory)
+    index2 = IndexWithData(clt, factory=factory)
+
+    yield [index1, index2]
+
+    # Tear down
+    index1.client.delete_index(index1.index_name)
+    index2.client.delete_index(index2.index_name)

tests/conftest.py

@@ -0,0 +1,16 @@
+from collections import namedtuple
+
+
+FakeResp = namedtuple('FakeResp', ['status_code', 'json'])
+
+
+class FakeSession:
+    def __init__(self, exp_headers, exp_params):
+        self.headers = exp_headers
+        self.params = exp_params
+
+    def request(self, path, meth, timeout, params, data, headers):
+        assert headers == self.headers
+        assert params == self.params
+
+        return FakeResp(status_code=200, json=lambda: '{}')

tests/fake_session.py

@@ -1,46 +1,125 @@
-# -*- coding: utf-8 -*-
-
-from __future__ import unicode_literals
-
 import os
 import time
 from random import randint
 
-from faker import Factory
-from algoliasearch.client import Client
+from algoliasearch import algoliasearch
+from faker import Faker
+
+
+def check_credentials():
+    credentials = [
+        'ALGOLIA_APPLICATION_ID',
+        'ALGOLIA_API_KEY',
+        'ALGOLIA_API_KEY_SEARCH',
+        'ALGOLIA_APPLICATION_ID_MCM',
+        'ALGOLIA_API_KEY_MCM'
+    ]
+
+    for credential in credentials:
+        if credential not in os.environ:
+            print('environement variable {} not defined')
+            assert False
 
-from collections import namedtuple
 
+def index_name():
+    name = 'algolia-python{}'.format(randint(1, 100000))
 
-class FakeData(object):
+    if 'TRAVIS' in os.environ:
+        name = '{}_travis-{}'.format(name, os.environ['TRAVIS_JOB_NUMBER'])
+
+    return name
+
+
+class Factory:
     def __init__(self):
-        self._fake = Factory.create('zh_CN')
-        self._fake.seed(4242)
-        self._generated_id = []
+        self._faker = Faker('zh_CN')
+        self.last_id = 0
 
-    def fake_contact(self, n=1):
+    def contacts(self, count=1):
         data = []
-        for i in range(n):
+        for _ in range(count):
             data.append({
-                'name': self._fake.name(),
-                'email': self._fake.email(),
-                'phone': self._fake.phone_number(),
-                'city': self._fake.city(),
-                'country': self._fake.country()
+                'name': self._faker.name(),
+                'email': self._faker.email(),
+                'phone': self._faker.phone_number(),
+                'city': self._faker.city(),
+                'country': self._faker.country()
             })
 
-        return data[0] if n == 1 else data
+        return data[0] if count == 1 else data
+
+    def with_objectids(self, ids):
+        if isinstance(ids, str) or isinstance(ids, int):
+            contacts = self.contacts()
+            contacts['objectID'] = ids
+            return contacts
+
+        contacts = self.contacts(len(ids))
+        for objid, contact in zip(ids, contacts):
+            contact['objectID'] = objid
+
+        return contacts
+
+    def objectids(self, n=1):
+        ids = [str(i) for i in range(self.last_id, self.last_id + n)]
+        self.last_id += n
+        return ids
+
+
+def create_client(
+        app_id_env='ALGOLIA_APPLICATION_ID',
+        api_key_env='ALGOLIA_API_KEY'
+):
+    if 'APPENGINE_RUNTIME' in os.environ:
+        from google.appengine.api import apiproxy_stub_map
+        from google.appengine.api import urlfetch_stub
+
+        apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
+        apiproxy_stub_map.apiproxy.RegisterStub(
+            'urlfetch', urlfetch_stub.URLFetchServiceStub()
+        )
 
-    def generate_id(self):
-        while True:
-            new_id = randint(0, 100000)
-            if new_id not in self._generated_id:
-                self._generated_id.append(new_id)
-                return str(new_id)
+    check_credentials()
+    try:
+        return algoliasearch.Client(
+            os.environ[app_id_env],
+            os.environ[api_key_env]
+        )
+    except:
+        raise RuntimeError(
+            "{} and {} environement variables must be set".format(
+                app_id_env, api_key_env
+            )
+        )
 
-def get_rule_stub(objectID='my-rule'):
+
+def create_index(clt):
+    name = index_name()
+    task = clt.delete_index(name)
+    idx = clt.init_index(name)
+    idx.wait_task(task['taskID'])
+
+    return idx
+
+
+class IndexWithData:
+    def __init__(self, clt, factory=None):
+        if factory is None:
+            factory = Factory()
+
+        self.data = factory.contacts(5)
+        self._index = create_index(clt)
+        task = self._index.add_objects(self.data)
+        self._index.wait_task(task['taskID'])
+        self.ids = task['objectIDs']
+
+    def __getattr__(self, name):
+        return getattr(self._index, name)
+
+
+def rule_stub(objid='my-rule'):
     return {
-        'objectID': objectID,
+        'objectID': objid,
         'condition': {
             'pattern': 'some text',
             'anchoring': 'is'
@@ -52,25 +131,9 @@ def get_rule_stub(objectID='my-rule'):
         }
     }
 
-def get_api_client():
-    if 'APPENGINE_RUNTIME' in os.environ:
-        from google.appengine.api import apiproxy_stub_map 
-        from google.appengine.api import urlfetch_stub
-        apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap() 
-        apiproxy_stub_map.apiproxy.RegisterStub('urlfetch',  urlfetch_stub.URLFetchServiceStub())
-    return Client(os.environ['ALGOLIA_APPLICATION_ID'],
-                  os.environ['ALGOLIA_API_KEY'])
-
-
-def safe_index_name(name):
-    if 'TRAVIS' not in os.environ:
-        return name
-    job = os.environ['TRAVIS_JOB_NUMBER']
-    return '{0}_travis-{1}'.format(name, job)
-
 
 def wait_key(index, key, block=None):
-    for i in range(60):
+    for _ in range(60):
         try:
             k = index.get_user_key_acl(key)
             if block is None or block(k):
@@ -82,24 +145,10 @@ def wait_key(index, key, block=None):
 
 
 def wait_missing_key(index, key):
-    for i in range(60):
+    for _ in range(60):
         try:
             index.get_user_key_acl(key)
             time.sleep(1)
         except:
-            # Not found.
+            # Not found
             return
-
-FakeResp = namedtuple('FakeResp', ['status_code', 'json'])
-
-class FakeSession(object):
-    def __init__(self, test_class, exp_headers, exp_params):
-        self.test_class = test_class
-        self.headers = exp_headers
-        self.params = exp_params
-
-    def request(self, path, meth, timeout, params, data, headers):
-        self.test_class.assertDictEqual(headers, self.headers)
-        self.test_class.assertDictEqual(params, self.params)
-
-        return FakeResp(status_code=200, json=lambda: '{}')