filled out api tests with the rest of the GET endpoints from the internal API interface

api/tests/api_tests.py

@@ -26,6 +26,7 @@ import requests
 import argparse
 import binascii
 import traceback
+import jsontokens
 
 from test import test_support
 from binascii import hexlify
@@ -36,7 +37,8 @@ import api.config
 
 from api.tests.resolver_tests import ResolverTestCase
 from api.tests.search_tests import SearchTestCase
-from blockstack_client import schemas 
+from blockstack_client import schemas
+import blockstack_client.storage
 import blockstack_client.config as blockstack_config
 import blockstack_client.config as blockstack_constants
 import blockstack_client.keys
@@ -66,7 +68,7 @@ class ForwardingClient:
         ret_obj.data = resp.text
         return ret_obj
     def post(self, endpoint, data, headers = {}):
-        resp = requests.post(self.base_url + endpoint, 
+        resp = requests.post(self.base_url + endpoint,
                              data = data, headers = headers)
         ret_obj = FakeResponseObj()
         ret_obj.status_code = resp.status_code
@@ -81,15 +83,19 @@ class APITestCase(unittest.TestCase):
         if not self.app:
             api.app.testing = True
             self.app = api.app.test_client()
-    def get_request(self, cls, endpoint, headers={}, status_code=200):
+    def get_request(self, endpoint, headers={}, status_code=200,
+                    no_json = False):
         t_start = time.time()
         resp = self.app.get(endpoint, headers = headers)
         t_end = time.time()
         print("\r{}get time: {}s".format("\t"*9, t_end - t_start))
         if not resp.status_code == status_code:
             print("{} => {} ".format(endpoint, resp.status_code))
-        
-        cls.assertEqual(resp.status_code, status_code)
+
+        self.assertEqual(resp.status_code, status_code)
+        if no_json:
+            return resp.data
+
         try:
             data = json.loads(resp.data)
             return data
@@ -98,15 +104,15 @@ class APITestCase(unittest.TestCase):
                 return {}
             raise e
 
-    def post_request(self, cls, endpoint, payload, headers={}, status_code=200):
+    def post_request(self, endpoint, payload, headers={}, status_code=200):
         t_start = time.time()
         resp = self.app.post(endpoint, data = json.dumps(payload), headers = headers)
         t_end = time.time()
         print("\r{}post time: {}s".format("\t"*9, t_end - t_start))
         if not resp.status_code == status_code:
             print("{} => {} ".format(endpoint, resp.status_code))
-        
-        cls.assertEqual(resp.status_code, status_code)
+
+        self.assertEqual(resp.status_code, status_code)
         try:
             data = json.loads(resp.data)
             return data
@@ -123,32 +129,26 @@ class InternalAPITestCase(APITestCase):
 def get_auth_header(key = API_PASSWORD):
     return {'Authorization' : 'bearer {}'.format(key)}
 
-def check_data(cls, data, required_keys=[], banned_keys=[]):
+def check_data(cls, data, required_keys={}):
     for k in required_keys:
-        cls.assertTrue(k in data)
-        for subkey in required_keys[k]:
-            cls.assertTrue(subkey in data[k])
-    for k in banned_keys:
-        if len(banned_keys[k]) is 0:
-            cls.assertTrue(k not in data)
-        else:
-            cls.assertTrue(k in data)
-            for subkey in banned_keys[k]:
-                cls.assertTrue(subkey not in data[k])
-
+        cls.assertIn(k, data)
+        if type(required_keys[k]) == dict:
+            check_data(cls, data[k], required_keys = required_keys[k])
+        if type(required_keys[k]) == str:
+            cls.assertRegexpMatches(data[k], required_keys[k])
+        if type(required_keys[k]) == int:
+            cls.assertGreaterEqual(data[k], required_keys[k])
 
 class PingTest(APITestCase):
     def test_found_user_lookup(self):
-        data = self.get_request(self, "/v1/ping",
+        data = self.get_request("/v1/ping",
                                 headers = {} , status_code=200)
-        
+
         self.assertTrue(data['status'] == 'alive')
 
 class AuthInternal(InternalAPITestCase):
     def test_get_and_use_session_token(self):
-        import jsontokens
-
-        privkey = ("a28ea1a6f11fb1c755b1d102990d64d6" + 
+        privkey = ("a28ea1a6f11fb1c755b1d102990d64d6" +
                    "b4468c10705bbcbdfca8bc4497cf8da8")
 
         auth_header = get_auth_header()
@@ -162,46 +162,67 @@ class AuthInternal(InternalAPITestCase):
         package = signer.sign(request, privkey)
 
         url = "/v1/auth?authRequest={}".format(package)
-        data = self.get_request(self, url, headers = auth_header, status_code=200)
+        data = self.get_request(url, headers = auth_header, status_code=200)
         self.assertIn('token', data)
         session = data['token']
 
         auth_header = get_auth_header(session)
-        data = self.get_request(self, '/v1/wallet/payment_address',
+        data = self.get_request('/v1/wallet/payment_address',
                                 headers = auth_header, status_code=200)
-        data = self.get_request(self, '/v1/users/muneeb.id',
+        data = self.get_request('/v1/users/muneeb.id',
                                 headers = auth_header, status_code=403)
-        
+
+class UsersInternal(InternalAPITestCase):
+    def test_get_users(self):
+        user = "muneeb.id"
+        data = self.get_request('/v1/users/{}'.format(user),
+                                headers = get_auth_header(), status_code=200)
+        to_check = {
+            "@type": True,
+        }
+        check_data(self, data, to_check)
 
 class LookupUsersTest(APITestCase):
-    def build_url(self, username):
-        return '/v1/names/{}'.format(username)
-
     def test_found_user_lookup(self):
-        usernames = 'muneeb.id'
-        data = self.get_request(self, self.build_url(usernames),
-                                headers = {}, status_code=200)
-        
+        base_url = '/v1/names/{}'
+        url = base_url.format('muneeb.id')
+        data = self.get_request(url, headers = {}, status_code=200)
+
         self.assertTrue(data['status'] == 'registered')
-        self.assertRegexpMatches(data['address'], schemas.OP_ADDRESS_PATTERN)
-        self.assertRegexpMatches(data['zonefile_hash'], schemas.OP_ZONEFILE_HASH_PATTERN)
-        self.assertRegexpMatches(data['last_txid'], schemas.OP_TXID_PATTERN)
 
-    def test_get_all_names(self):
-        data = test_get_request(self, "/v1/names?page=0",
-                                headers = {} , status_code=200)
-        self.assertEqual(len(data), 100, "Paginated name length != 100")
-        data = test_get_request(self, "/v1/names",
-                                headers = {} , status_code=401)
-        data = test_get_request(self, "/v1/names?page=10000",
-                                headers = {} , status_code=200)
+        to_check = {'address': schemas.OP_ADDRESS_PATTERN,
+                    'zonefile_hash' : schemas.OP_ZONEFILE_HASH_PATTERN,
+                    'last_txid' : schemas.OP_TXID_PATTERN}
+        check_data(self, data, to_check)
 
-    def test_user_not_formatted(self):
-        usernames = 'muneeb'
-        data = self.get_request(self, self.build_url(usernames),
-                                headers = {}, status_code=500)
+        url = base_url.format('muneeb')
+        data = self.get_request(url, headers = {}, status_code=500)
         self.assertTrue(data['error'] == 'Failed to lookup name')
 
+    def test_get_all_names(self):
+        data = self.get_request("/v1/names?page=0",
+                                headers = {} , status_code=200)
+        self.assertEqual(len(data), 100, "Paginated name length != 100")
+        data = self.get_request("/v1/names",
+                                headers = {} , status_code=401)
+        data = self.get_request("/v1/names?page=10000",
+                                headers = {} , status_code=200)
+
+class Zonefiles(APITestCase):
+    def test_get_zonefile(self):
+        zf_url = '/v1/names/{}/zonefile'
+        zf_hash_url = '/v1/names/{}/zonefile/{}'
+        user = 'muneeb.id'
+
+        zf_data = self.get_request(zf_url.format(user),
+                                   headers = {}, status_code = 200)
+        self.assertIn('zonefile', zf_data)
+
+        zf_hash = blockstack_client.storage.get_zonefile_data_hash(zf_data['zonefile'])
+        zf_data_historic_lookup = self.get_request(zf_hash_url.format(user, zf_hash),
+                                                   headers = {}, status_code = 200)
+        self.assertEqual(zf_data_historic_lookup['zonefile'],
+                         zf_data['zonefile'])
 
 class NameHistoryTest(APITestCase):
     def build_url(self, username):
@@ -215,9 +236,9 @@ class NameHistoryTest(APITestCase):
 
     def test_found_user_lookup(self):
         usernames = 'muneeb.id'
-        data = self.get_request(self, self.build_url(usernames),
+        data = self.get_request(self.build_url(usernames),
                                 headers = {}, status_code=200)
-        
+
         self.assertTrue(len(data.keys()) > 2)
         for block_key, block_data in data.items():
             self.check_history_block(block_data)
@@ -231,32 +252,49 @@ class NamesOwnedTest(APITestCase):
                           "16EMaNw3pkn3v6f2BgnSSs53zAKH4Q8YJg"]
         names_to_check = ["muneeb.id", "judecn.id"]
         for addr, name in zip(addrs_to_check, names_to_check):
-            data = self.get_request(self, self.build_url(addr),
+            data = self.get_request(self.build_url(addr),
                                     headers = {}, status_code = 200)
             self.assertTrue(len(data["names"]) > 0)
             self.assertIn(name, data["names"])
 
 class NamespaceTest(APITestCase):
     def test_id_space(self):
-        data = self.get_request(self, "/v1/namespaces",
-                                headers = {} , status_code=200)        
+        data = self.get_request("/v1/namespaces",
+                                headers = {} , status_code=200)
         self.assertIn('id', data)
+
+        data = self.get_request('/v1/namespaces/id', headers = {}, status_code = 200)
+        to_check = {
+            "address": schemas.OP_ADDRESS_PATTERN,
+            "block_number": 0,
+            "history": True,
+            "namespace_id": True,
+            "op": True,
+            "op_fee": 0,
+            "preorder_hash": schemas.OP_HEX_PATTERN,
+            "ready_block": 0,
+            "reveal_block": 0,
+            "sender": schemas.OP_HEX_PATTERN,
+            "sender_pubkey": schemas.OP_PUBKEY_PATTERN,
+            "txid": schemas.OP_TXID_PATTERN
+        }
+
+        check_data(self, data, to_check)
+
     def test_id_space_names(self):
-        data = test_get_request(self, "/v1/namespaces/id/names?page=0",
+        data = self.get_request("/v1/namespaces/id/names?page=0",
                                 headers = {} , status_code=200)
         self.assertEqual(len(data), 100, "Paginated name length != 100")
-        data = test_get_request(self, "/v1/namespaces/id/names",
+        data = self.get_request("/v1/namespaces/id/names",
                                 headers = {} , status_code=401)
-        
 
 
-class NamepriceTest(APITestCase):
-    def price_url(self, name):
-        return "/v1/prices/names/{}".format(name)
-    
+
+class Prices(APITestCase):
     def test_id_price(self):
-        data = self.get_request(self, self.price_url("muneeb.id"),
-                                headers = {} , status_code=200)        
+        price_url = "/v1/prices/names/{}"
+        data = self.get_request(price_url.format("muneeb.id"),
+                                headers = {} , status_code=200)
         json_keys = data.keys()
         self.assertIn('name_price', json_keys)
         self.assertIn('preorder_tx_fee', json_keys)
@@ -265,6 +303,10 @@ class NamepriceTest(APITestCase):
         self.assertIn('total_tx_fees', json_keys)
         self.assertIn('update_tx_fee', json_keys)
 
+    def test_ns_price(self):
+        data = self.get_request("/v1/prices/namespaces/id",
+                                headers = {} , status_code=200)
+        check_data(self, data, {'satoshis':0})
 
 class SearchAPITest(APITestCase):
     def search_url(self, q):
@@ -274,8 +316,8 @@ class SearchAPITest(APITestCase):
         u = "muneeb"
         original = api.config.SEARCH_API_ENDPOINT_ENABLED
         api.config.SEARCH_API_ENDPOINT_ENABLED = False
-        
-        data = self.get_request(self, self.search_url(u),
+
+        data = self.get_request(self.search_url(u),
                                 headers = {}, status_code=200)
 
         self.assertTrue(len(data['results']) > 0)
@@ -283,13 +325,13 @@ class SearchAPITest(APITestCase):
         self.assertIn("profile", data['results'][0].keys())
 
         api.config.SEARCH_API_ENDPOINT_ENABLED = original
-        
+
     def test_search_server(self):
         u = "muneeb"
         if not api.config.SEARCH_API_ENDPOINT_ENABLED:
             print("skipping search server test")
             return
-        data = self.get_request(self, self.search_url(u),
+        data = self.get_request(self.search_url(u),
                                 headers = {}, status_code=200)
 
         self.assertTrue(len(data['results']) > 0)
@@ -300,41 +342,50 @@ class TestAPILandingPageExamples(APITestCase):
     def test_endpoints(self):
         from api.utils import get_api_calls
         current_dir = os.path.abspath(os.path.dirname(__file__))
-        api_endpoints = [ call['tryit_pathname'] 
+        api_endpoints = [ call['tryit_pathname']
                           for call in get_api_calls(current_dir + '/../api_v1.md')
                           if (not ("private" in call and call["private"].lower().startswith("t")))
                           and 'tryit_pathname' in call ]
         print("")
         for url in api_endpoints:
             print("\tTesting API example: {}".format(url))
-            self.get_request(self, url, headers = {}, status_code=200)
+            self.get_request(url, headers = {}, status_code=200)
 
 class BlockChains(APITestCase):
     def test_consensus(self):
-        data = self.get_request(self, "/v1/blockchains/bitcoin/consensus",
-                                headers = {} , status_code=200)        
+        data = self.get_request("/v1/blockchains/bitcoin/consensus",
+                                headers = {} , status_code=200)
         self.assertRegexpMatches(data['consensus_hash'], schemas.OP_CONSENSUS_HASH_PATTERN)
     def test_name_history(self):
-        data = self.get_request(self, "/v1/blockchains/bitcoin/names/muneeb.id/history",
-                                headers = {} , status_code=405)        
+        data = self.get_request("/v1/blockchains/bitcoin/names/muneeb.id/history",
+                                headers = {} , status_code=405)
     def test_names_pending(self):
-        data = self.get_request(self, "/v1/blockchains/bitcoin/pending",
+        data = self.get_request("/v1/blockchains/bitcoin/pending",
                                 headers = {} , status_code=200)
         self.assertIn("queues", data)
     def test_operations(self):
-        data = self.get_request(self, "/v1/blockchains/bitcoin/operations",
+        data = self.get_request("/v1/blockchains/bitcoin/operations/462453",
                                 headers = {} , status_code=200)
 
+        to_check = {"address" : schemas.OP_ADDRESS_PATTERN,
+                    "block_number" : 0,
+                    "consensus_hash": schemas.OP_CONSENSUS_HASH_PATTERN,
+                    "first_registered": 0,
+                    "history" : True,
+                    "op" : True,
+                    "txid": schemas.OP_HEX_PATTERN,
+                    "value_hash": schemas.OP_HEX_PATTERN}
+        check_data(self, data[0], to_check)
 
 class BlockChainsInternal(InternalAPITestCase):
     def test_unspents(self):
         url = "/v1/blockchains/bitcoin/{}/unspent".format(DEFAULT_WALLET_ADDRESS)
-        self.get_request(self, url, headers = {}, status_code = 403)
-        data = self.get_request(self, url, headers = get_auth_header(), status_code = 200)
-        
+        self.get_request(url, headers = {}, status_code = 403)
+        data = self.get_request(url, headers = get_auth_header(), status_code = 200)
+
         self.assertTrue(len(data) >= 1)
         data = data[0]
-        
+
         self.assertTrue(data['confirmations'] >= 0)
         self.assertRegexpMatches(data['out_script'], schemas.OP_HEX_PATTERN)
         self.assertRegexpMatches(data['outpoint']['hash'], schemas.OP_HEX_PATTERN)
@@ -342,14 +393,58 @@ class BlockChainsInternal(InternalAPITestCase):
         self.assertTrue(data['value'] >= 0)
     def test_txs(self):
         url = "/v1/blockchains/bitcoin/txs".format(DEFAULT_WALLET_ADDRESS)
-        self.post_request(self, url, payload = {}, headers = {}, status_code = 403)
-        self.post_request(self, url, payload = {}, headers = get_auth_header(), status_code = 401)
-        
+        self.post_request(url, payload = {}, headers = {}, status_code = 403)
+        self.post_request(url, payload = {}, headers = get_auth_header(), status_code = 401)
+
+class WalletInternal(InternalAPITestCase):
+    def test_addresses(self):
+        for endpoint in ['payment_address', 'owner_address']:
+            data = self.get_request("/v1/wallet/{}".format(endpoint),
+                                    headers = get_auth_header(), status_code = 200)
+            self.assertRegexpMatches(data['address'], schemas.OP_ADDRESS_PATTERN)
+        data = self.get_request("/v1/wallet/data_pubkey",
+                                headers = get_auth_header(), status_code = 200)
+        self.assertRegexpMatches(data['public_key'], schemas.OP_PUBKEY_PATTERN)
+    def test_balance(self):
+        data = self.get_request("/v1/wallet/balance",
+                                headers = get_auth_header(), status_code = 200)
+        to_check = {'balance' : { 'bitcoin' : 0, 'satoshis' : 0 } }
+        check_data(self, data, to_check)
+    def test_keys(self):
+        data = self.get_request("/v1/wallet/keys",
+                                headers = get_auth_header(), status_code = 200)
+        to_check = {
+            "data_privkey": schemas.OP_HEX_PATTERN,
+            "data_pubkey": schemas.OP_PUBKEY_PATTERN,
+            "owner_address": schemas.OP_ADDRESS_PATTERN,
+            "owner_privkey": True,
+            "payment_address": schemas.OP_ADDRESS_PATTERN,
+            "payment_privkey": True,
+        }
+
+        check_data(self, data, to_check)
+
+class NodeInternal(InternalAPITestCase):
+    def test_registrar(self):
+        self.get_request("/v1/node/registrar/state", headers = get_auth_header(),
+                         status_code = 200)
+    def test_get_log(self):
+        self.get_request("/v1/node/log", headers = get_auth_header(),
+                         status_code = 200, no_json = True)
+    def test_config(self):
+        data = self.get_request("/v1/node/config", headers = get_auth_header(),
+                                status_code = 200)
+        to_check = { "bitcoind": True,
+                     "blockchain-reader": True,
+                     "blockchain-writer": True,
+                     "blockstack-client": True }
+
+        check_data(self, data, to_check)
 
 def test_main(args = []):
     test_classes = [PingTest, LookupUsersTest, NamespaceTest, BlockChains, TestAPILandingPageExamples,
-                    NamepriceTest, NamesOwnedTest, NameHistoryTest, SearchAPITest, 
-                    AuthInternal, BlockChainsInternal]
+                    Prices, NamesOwnedTest, NameHistoryTest, SearchAPITest,
+                    AuthInternal, BlockChainsInternal, Zonefiles, WalletInternal, NodeInternal]
     test_classes += [ResolverTestCase]
     if api.config.SEARCH_API_ENDPOINT_ENABLED:
         test_classes += [SearchTestCase]