First unit tests for coingecko wrappers

.pytest_cache/README.md

@@ -0,0 +1,8 @@
+# pytest cache directory #
+
+This directory contains data from the pytest's cache plugin,
+which provides the `--lf` and `--ff` options, as well as the `cache` fixture.
+
+**Do not** commit this to version control.
+
+See [the docs](https://docs.pytest.org/en/latest/cache.html) for more information.

.pytest_cache/v/cache/lastfailed

@@ -0,0 +1 @@
+{}

.pytest_cache/v/cache/nodeids

@@ -0,0 +1,12 @@
+[
+  "tests/test_api.py::TestWrapper::test_failed_get_coin_by_id",
+  "tests/test_api.py::TestWrapper::test_failed_get_coins",
+  "tests/test_api.py::TestWrapper::test_failed_get_coins_list",
+  "tests/test_api.py::TestWrapper::test_failed_get_coins_markets",
+  "tests/test_api.py::TestWrapper::test_failed_ping",
+  "tests/test_api.py::TestWrapper::test_get_coin_by_id",
+  "tests/test_api.py::TestWrapper::test_get_coins",
+  "tests/test_api.py::TestWrapper::test_get_coins_list",
+  "tests/test_api.py::TestWrapper::test_get_coins_markets",
+  "tests/test_api.py::TestWrapper::test_ping"
+]

build/lib/pycoingecko/__init__.py

@@ -0,0 +1 @@
+from .api import CoinGeckoAPI 

build/lib/pycoingecko/api.py

@@ -0,0 +1,132 @@
+import json
+import requests
+
+class CoinGeckoAPI:
+
+    __API_URL_BASE = 'https://api.coingecko.com/api/v3/'
+
+    def __init__(self, api_base_url = __API_URL_BASE):
+        self.api_base_url = api_base_url
+        self.request_timeout = 120
+
+
+    def __request(self, url):
+        #print(url)
+        try:
+            response = requests.get(url, timeout = self.request_timeout)
+            response.raise_for_status()
+            content = json.loads(response.content.decode('utf-8'))
+            return content
+        except Exception as e:
+            raise
+
+
+    def __api_url_params(self, api_url, params):
+        if params:
+            api_url += '?'
+            for key, value in params.items():
+                api_url += "{0}={1}&".format(key, value)
+            api_url = api_url[:-1]
+        return api_url
+
+
+    #---------- PING ----------#
+    def ping(self):
+        """Check API server status"""
+
+        api_url = '{0}ping'.format(self.api_base_url)
+        return self.__request(api_url)
+
+
+    #---------- COINS ----------#
+    def get_coins(self, **kwargs):
+        """List all coins with data (name, price, market, developer, community, etc)"""
+
+        api_url = '{0}coins'.format(self.api_base_url)
+        #['order', 'per_page', 'page', 'localization'] 
+        api_url = self.__api_url_params(api_url, kwargs)
+
+        return self.__request(api_url)
+
+
+    def get_coins_list(self):
+        """List all supported coins id, name and symbol (no pagination required)"""
+
+        api_url = '{0}coins/list'.format(self.api_base_url)
+
+        return self.__request(api_url)
+
+
+    def get_coins_markets(self, vs_currency, **kwargs):
+        """List all supported coins price, market cap, volume, and market related data (no pagination required)"""
+
+        kwargs['vs_currency'] = vs_currency
+
+        api_url = '{0}coins/markets'.format(self.api_base_url)
+        api_url = self.__api_url_params(api_url, kwargs)
+
+        return self.__request(api_url)
+
+
+    def get_coin_by_id(self, id, **kwargs):
+        """Get current data (name, price, market, ... including exchange tickers) for a coin"""
+
+        api_url = '{0}coins/{1}/'.format(self.api_base_url, id)
+        api_url = self.__api_url_params(api_url, kwargs)
+
+        return self.__request(api_url)
+
+
+    def get_coin_history_by_id(self, id, date, **kwargs):
+        """Get historical data (name, price, market, stats) at a given date for a coin"""
+
+        kwargs['date'] = date 
+
+        api_url = '{0}coins/{1}/history'.format(self.api_base_url, id)
+        api_url = self.__api_url_params(api_url, kwargs)
+
+        return self.__request(api_url)
+
+
+    def get_coin_market_chart_by_id(self, id, vs_currency, days):
+        """Get historical market data include price, market cap, and 24h volume (granularity auto)"""
+
+        api_url = '{0}coins/{1}/market_chart?vs_currency={2}&days={3}'.format(self.api_base_url, id, vs_currency, days)
+
+        return self.__request(api_url)
+
+
+    #---------- EXCHANGES ----------#
+    def get_exchanges_list(self):
+        """List all exchanges"""
+
+        api_url = '{0}exchanges/list'.format(self.api_base_url)
+
+        return self.__request(api_url)
+
+
+    def get_exchanges_by_id(self, id):
+        """Get exchange volume in BTC and tickers"""
+
+        api_url = '{0}exchanges/{1}'.format(self.api_base_url, id)
+
+        return self.__request(api_url)
+
+
+    #---------- EXCHANGE-RATES ----------#
+    def get_exchange_rates(self):
+        """Get BTC-to-Currency exchange rates"""
+
+        api_url = '{0}exchange_rates'.format(self.api_base_url)
+
+        return self.__request(api_url)
+
+
+    #---------- GLOBAL ----------#
+    def get_global(self):
+        """Get cryptocurrency global data"""
+
+        api_url = '{0}global'.format(self.api_base_url)
+
+        return self.__request(api_url)['data']
+

pycoingecko/api.py

@@ -43,9 +43,9 @@ def get_coins(self, **kwargs):
         """List all coins with data (name, price, market, developer, community, etc)"""
 
         api_url = '{0}coins'.format(self.api_base_url)
-        #['order', 'per_page', 'page', 'localization'] 
+        #['order', 'per_page', 'page', 'localization']
         api_url = self.__api_url_params(api_url, kwargs)
-        
+
         return self.__request(api_url)
 
 
@@ -70,7 +70,7 @@ def get_coins_markets(self, vs_currency, **kwargs):
 
     def get_coin_by_id(self, id, **kwargs):
         """Get current data (name, price, market, ... including exchange tickers) for a coin"""
-        
+
         api_url = '{0}coins/{1}/'.format(self.api_base_url, id)
         api_url = self.__api_url_params(api_url, kwargs)
 
@@ -79,9 +79,9 @@ def get_coin_by_id(self, id, **kwargs):
 
     def get_coin_history_by_id(self, id, date, **kwargs):
         """Get historical data (name, price, market, stats) at a given date for a coin"""
-        
-        kwargs['date'] = date 
-        
+
+        kwargs['date'] = date
+
         api_url = '{0}coins/{1}/history'.format(self.api_base_url, id)
         api_url = self.__api_url_params(api_url, kwargs)
 
@@ -129,4 +129,3 @@ def get_global(self):
         api_url = '{0}global'.format(self.api_base_url)
 
         return self.__request(api_url)['data']
-

tests/test_api.py

@@ -0,0 +1,134 @@
+import pytest
+import responses
+import unittest
+import unittest.mock as mock
+
+from pycoingecko import CoinGeckoAPI
+from requests.exceptions import HTTPError
+
+class TestWrapper(unittest.TestCase):
+
+    @responses.activate
+    def test_failed_ping(self):
+        # Arrange
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/ping',
+                          status = 404)
+        exception = HTTPError("HTTP Error")
+
+        # Act
+        with pytest.raises(HTTPError) as HE:
+            CoinGeckoAPI().ping()
+
+    @responses.activate
+    def test_ping(self):
+        # Arrange
+        ping_json = { 'gecko_says':'(V3) To the Moon!' }
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/ping',
+                          json = ping_json, status = 200)
+
+        # Act
+        response = CoinGeckoAPI().ping()
+
+        ## Assert
+        assert response == ping_json
+
+    @responses.activate
+    def test_failed_get_coins(self):
+        # Arrange
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins',
+                          status = 404)
+        exception = HTTPError("HTTP Error")
+
+        # Act
+        with pytest.raises(HTTPError) as HE:
+            CoinGeckoAPI().get_coins()
+
+    @responses.activate
+    def test_get_coins(self):
+        # Arrange
+        coins_json_sample = [ { "id": "bitcoin", "symbol": "btc", "name": "Bitcoin", "localization": { "en": "Bitcoin", "es": "Bitcoin", "de": "Bitcoin", "nl": "Bitcoin", "pt": "Bitcoin", "fr": "Bitcoin", "it": "Bitcoin", "hu": "Bitcoin", "ro": "Bitcoin", "sv": "Bitcoin", "pl": "Bitcoin", "id": "Bitcoin", "zh": "Bitcoin", "zh-tw": "Bitcoin", "ja": "Bitcoin", "ko": "Bitcoin", "ru": "Bitcoin", "ar": "Bitcoin", "th": "Bitcoin", "vi": "Bitcoin", "tr": "Bitcoin" }, "image": { "thumb": "https://assets.coingecko.com/coins/images/1/thumb/bitcoin.png?1510040391", "small": "https://assets.coingecko.com/coins/images/1/small/bitcoin.png?1510040391", "large": "https://assets.coingecko.com/coins/images/1/large/bitcoin.png?1510040391" } } ]
+
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins',
+                          json = coins_json_sample, status = 200)
+
+        # Act
+        response = CoinGeckoAPI().get_coins()
+
+        ## Assert
+        assert response == coins_json_sample
+
+    @responses.activate
+    def test_failed_get_coins_list(self):
+        # Arrange
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/list',
+                          status = 404)
+        exception = HTTPError("HTTP Error")
+
+        # Act
+        with pytest.raises(HTTPError) as HE:
+            CoinGeckoAPI().get_coins_list()
+
+    @responses.activate
+    def test_get_coins_list(self):
+        # Arrange
+        coins_json_sample = [ { "id": "bitcoin", "symbol": "btc", "name": "Bitcoin" }, { "id": "litecoin", "symbol": "ltc", "name": "Litecoin" }, { "id": "auroracoin", "symbol": "aur", "name": "Auroracoin" }, { "id": "peercoin", "symbol": "ppc", "name": "Peercoin" }, { "id": "dogecoin", "symbol": "doge", "name": "Dogecoin" }, { "id": "nxt", "symbol": "nxt", "name": "NXT" }, { "id": "omni", "symbol": "omni", "name": "Omni (Mastercoin)" } ]
+
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/list',
+                          json = coins_json_sample, status = 200)
+
+        # Act
+        response = CoinGeckoAPI().get_coins_list()
+
+        ## Assert
+        assert response == coins_json_sample
+
+    @responses.activate
+    def test_failed_get_coins_markets(self):
+        # Arrange
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/markets?vs_currency=usd',
+                          status = 404)
+        exception = HTTPError("HTTP Error")
+
+        # Act
+        with pytest.raises(HTTPError) as HE:
+            CoinGeckoAPI().get_coins_markets('usd')
+
+    @responses.activate
+    def test_get_coins_markets(self):
+        # Arrange
+        markets_json_sample = [ { "id": "bitcoin", "symbol": "btc", "name": "Bitcoin", "image": "https://assets.coingecko.com/coins/images/1/large/bitcoin.png?1510040391", "current_price": 7015.11823787848, "market_cap": 120934444800.105, "market_cap_rank": 1, "total_volume": 6121170828.21792, "high_24h": 7054.21193531031, "low_24h": 6668.29100755648, "price_change_24h": "299.72373285508", "price_change_percentage_24h": "4.46323343521924", "market_cap_change_24h": "5197755386.983", "market_cap_change_percentage_24h": "4.4910178555649", "circulating_supply": "17236100.0", "ath": 19665.3949272416, "ath_change_percentage": -64.2200698307594, "ath_date": "2017-12-16T00:00:00.000Z", "roi": 0, "last_updated": "2018-08-28T12:12:53.390Z" } ]
+
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/markets?vs_currency=usd',
+                          json = markets_json_sample, status = 200)
+
+        # Act
+        response = CoinGeckoAPI().get_coins_markets('usd')
+
+        ## Assert
+        assert response == markets_json_sample
+
+    @responses.activate
+    def test_failed_get_coin_by_id(self):
+        # Arrange
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/bitcoin/',
+                          status = 404)
+        exception = HTTPError("HTTP Error")
+
+        # Act
+        with pytest.raises(HTTPError) as HE:
+            CoinGeckoAPI().get_coin_by_id('bitcoin')
+
+
+    @responses.activate
+    def test_get_coin_by_id(self):
+        # Arrange
+        bitcoin_json_sample = { "id": "bitcoin", "symbol": "btc", "name": "Bitcoin", "categories": [ "Cryptocurrency" ], "localization": { "en": "Bitcoin", "es": "Bitcoin", "de": "Bitcoin", "nl": "Bitcoin", "pt": "Bitcoin", "fr": "Bitcoin", "it": "Bitcoin", "hu": "Bitcoin", "ro": "Bitcoin", "sv": "Bitcoin", "pl": "Bitcoin", "id": "Bitcoin", "zh": "比特币", "zh-tw": "比特幣", "ja": "ビットコイン", "ko": "비트코인", "ru": "биткоина", "ar": "بيتكوين", "th": "บิตคอยน์", "vi": "Bitcoin", "tr": "Bitcoin"}}
+
+        responses.add(responses.GET, 'https://api.coingecko.com/api/v3/coins/bitcoin/',
+                          json = bitcoin_json_sample, status = 200)
+
+        # Act
+        response = CoinGeckoAPI().get_coin_by_id('bitcoin')
+
+        ## Assert
+        assert response == bitcoin_json_sample