Merge pull request #15 from nym/candidates-and-watchlists

Update for Candidates & Watchlist v4.0 API Support

Author: alainmeier

blockscore/api/__init__.py

@@ -2,3 +2,4 @@
 from blockscore.api import people
 from blockscore.api import question_sets
 from blockscore.api import companies
+from blockscore.api import watchlists

blockscore/api/watchlists.py

@@ -0,0 +1,61 @@
+WATCHLISTS_CANDIDATE_URI = '/candidates'
+WATCHLISTS_SEARCH_URI = '/watchlists'
+
+
+class Watchlists():
+  def __init__(self, client):
+    self.client = client
+
+  @staticmethod
+  def _create_body(date_of_birth=None, identification=None,
+           name=None, address=None, note=None):
+    body = {
+      'date_of_birth': date_of_birth,
+      'note': note
+    }
+    if identification:
+      body['ssn'] = identification.get('ssn')
+      body['passport'] = identification.get('passport')
+    if name:
+      body['name_first'] = name.get('first')
+      body['name_middle'] = name.get('middle')
+      body['name_last'] = name.get('last')
+    if address:
+      body['address_street1'] = address.get('street1')
+      body['address_street2'] = address.get('street2')
+      body['address_city'] = address.get('city')
+      body['address_subdivision'] = address.get('state')
+      body['address_postal_code'] = address.get('postal_code')
+      body['address_country_code'] = address.get('country_code')
+    return body
+
+  def create(self, **kwargs):
+    body = self._create_body(**kwargs)
+    return self.client.post(WATCHLISTS_CANDIDATE_URI, body)
+
+  def edit(self, watchlists_candidate_id, **kwargs):
+    body = self._create_body(**kwargs)
+    return self.client.patch("%s/%s" % (WATCHLISTS_CANDIDATE_URI,
+                      watchlists_candidate_id), body)
+
+  def retrieve(self, watchlists_candidate_id):
+    return self.client.get("%s/%s" % (WATCHLISTS_CANDIDATE_URI,
+                      watchlists_candidate_id))
+
+  def delete(self, watchlists_candidate_id):
+    return self.client.delete("%s/%s" % (WATCHLISTS_CANDIDATE_URI,
+                       watchlists_candidate_id))
+
+  def list(self, count=None, offset=None):
+    body = {
+      'count': count,
+      'offset': offset
+    }
+    return self.client.get(WATCHLISTS_CANDIDATE_URI, body)
+
+  def search(self, watchlists_candidate_id, match_type=None):
+    body = {
+      'candidate_id': watchlists_candidate_id,
+      'match_type': match_type
+    }
+    return self.client.post(WATCHLISTS_SEARCH_URI, body)

blockscore/client.py

@@ -4,6 +4,8 @@
 from blockscore.api.people import People
 from blockscore.api.question_sets import QuestionSets
 from blockscore.api.companies import Companies
+#from .api.verification import Verification
+from blockscore.api.watchlists import Watchlists
 
 class Client():
 
@@ -12,4 +14,6 @@ def __init__(self, auth = {}, options = {}):
     self.people = People(self.http_client)
     self.question_sets = QuestionSets(self.http_client)
     self.companies = Companies(self.http_client)
+    #self.verification = Verification(self.http_client)
+    self.watchlists = Watchlists(self.http_client)
 

test/__init__.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python
+if __name__ == '__main__' and __package__ is None:
+  from os import sys, path
+  sys.path.append(path.dirname(path.dirname(path.abspath(__file__))))
+import blockscore
+import unittest
+import os, sys
+from unittest import skip
+import time
+
+class TestWatchlistsAPI(unittest.TestCase):
+  def setUp(self):
+    try:
+      self.client = blockscore.Client({'api_key': os.environ['BLOCKSCORE_API']})
+    except KeyError:
+      sys.stderr.write("To run tests, you must have a BLOCKSCORE_API environment variable with a test api key\n")
+      sys.exit(2)
+
+    self.test_identity = {
+      'date_of_birth': '1980-01-01',
+      'identification': {
+        'ssn': '0000',
+        'passport': '12315235'
+      },
+      'name': {
+        'first': 'john',
+        'middle': 'a',
+        'last': 'doe'
+      },
+      'address': {
+        'street1': '1 Infinite Loop',
+        'street2': 'Floor 3',
+        'city': 'Palo Alto',
+        'state': 'ca',
+        'postal_code': '94309',
+        'country_code': 'us'
+      },
+      'note': 'test'
+    }
+
+  def test_create_body(self):
+    body = self.client.watchlists._create_body(**self.test_identity)
+    self.assertDictEqual({
+      'ssn': self.test_identity['identification']['ssn'],
+      'date_of_birth': self.test_identity['date_of_birth'],
+      'passport': self.test_identity['identification']['passport'],
+      'name_first': self.test_identity['name']['first'],
+      'name_middle': self.test_identity['name']['middle'],
+      'name_last': self.test_identity['name']['last'],
+      'address_street1': self.test_identity['address']['street1'],
+      'address_street2': self.test_identity['address']['street2'],
+      'address_city': self.test_identity['address']['city'],
+      'address_subdivision': self.test_identity['address']['state'],
+      'address_postal_code': self.test_identity['address']['postal_code'],
+      'address_country_code': self.test_identity['address']['country_code'],
+      'note': self.test_identity['note']
+    }, body)
+
+  def test_create_candidate(self):
+    candidate = self.client.watchlists.create(**self.test_identity)
+    expected_body = self.client.watchlists._create_body(**self.test_identity)
+    self.assertDictContainsSubset(expected_body, candidate.body)
+
+  def test_edit_candidate(self):
+    candidate_id = self.client.watchlists.create(**self.test_identity).body['id']
+    candidate = self.client.watchlists.edit(candidate_id, date_of_birth='1980-01-02')
+    self.test_identity['date_of_birth'] = '1980-01-02'
+    expected_body = self.client.watchlists._create_body(**self.test_identity)
+    expected_body['id'] = candidate_id
+    self.assertDictContainsSubset(expected_body, candidate.body)
+
+  def test_retrieve_candidate(self):
+    candidate_id = self.client.watchlists.create(**self.test_identity).body['id']
+    candidate = self.client.watchlists.retrieve(candidate_id)
+    expected_body = self.client.watchlists._create_body(**self.test_identity)
+    expected_body['id'] = candidate_id
+    self.assertDictContainsSubset(expected_body, candidate.body)
+
+  def test_delete_candidate(self):
+    candidate_id = self.client.watchlists.create(**self.test_identity).body['id']
+    candidate = self.client.watchlists.delete(candidate_id)
+    expected_body = self.client.watchlists._create_body(**self.test_identity)
+    expected_body['id'] = candidate_id
+    expected_body['deleted'] = True
+    self.assertDictContainsSubset(expected_body, candidate.body)
+
+  @skip('These tests do not pass consistently because of eventual consistency of candidate list')
+  def test_list_candidates_with_params(self):
+    candidate = self.client.watchlists.create()
+    candidate_list = self.client.watchlists.list(count=1)
+    self.assertListEqual([candidate.body], candidate_list.body)
+    self.client.watchlists.create()
+    candidate_list = self.client.watchlists.list(count=1, offset=1)
+    self.assertListEqual([candidate.body], candidate_list.body)
+
+  #@skip('These tests do not pass consistently because of eventual consistency of candidate list')
+  def test_list_candidates_all(self):
+    candidate = self.client.watchlists.create(**self.test_identity)
+    expected_body = self.client.watchlists._create_body(**self.test_identity)
+    candidate_id = candidate.body['id']
+    time.sleep(5)
+    candidate_list = self.client.watchlists.list()
+    match = {}
+    for c in candidate_list.body['data']:
+      if c[u'id'] == candidate_id:
+        match = c
+    self.assertDictContainsSubset(expected_body, match)
+
+  def test_search(self):
+    candidate_id = self.client.watchlists.create().body['id']
+    response = self.client.watchlists.search(candidate_id)
+    self.assertIn('searched_lists', response.body)
+    self.assertIn('matches', response.body)