Making Adapter Code More Testable
Part of my Vigilant Crypto Snatch project are the marketplaces. The most popular is the Kraken one. I use the krakenex Python library to interface with that. In the UML diagram you can see how the KrakenexMarketplace
directly uses the krakenex
library:
In order to instantiate the KrakenexMarketplace
, one needs to supply the API key. One does this via the KrakenConfig
, which is just a data class that contains the API key and associated secret.
class KrakenexMarketplace(Marketplace): def __init__(self, config: KrakenConfig): self.handle = krakenex.API(config.key, config.secret) self.withdrawal_config = config.withdrawal self.prefer_fee_in_base_currency = config.prefer_fee_in_base_currency
This means that one cannot test this class without using an actual API key. But it still contains code that I'd like to test. For instance the get_spot_price()
method is rather simple, so there is not much that could go wrong:
def get_spot_price(self, coin: str, fiat: str, now: datetime.datetime) -> Price: answer = self.handle.query_public( "Ticker", {"pair": f"{map_normal_to_kraken(coin)}{fiat}"} ) raise_error(answer, TickerError) close = float(list(answer["result"].values())[0]["c"][0]) logger.debug(f"Retrieved {close} for {fiat}/{coin} from Krakenex.") price = Price(timestamp=now, last=close, coin=coin, fiat=fiat) return price
Yet there are a few things which are untested. Especially I am not sure whether the error conditions are properly handled.
It gets worse with the withdrawal()
method:
def withdrawal(self, coin: str, volume: float) -> None: if coin not in self.withdrawal_config: logger.debug(f"No withdrawal config for {coin}.") return if volume == 0: return target = self.withdrawal_config[coin].target fee = self.get_withdrawal_fee(coin, volume) if fee / volume <= self.withdrawal_config[coin].fee_limit_percent / 100: logger.info( f"Trying to withdraw {volume} {coin} as fee is just {fee} {coin} and below limit." ) answer = self.handle.query_private( "Withdraw", { "asset": map_normal_to_kraken(coin), "amount": volume, "key": target, }, ) raise_error(answer, WithdrawalError) else: logger.debug( f"Not withdrawing {volume} {coin} as fee is {fee} {coin} and above limit." )
This even calls the get_withdrawal_fee()
method, which issues another API call. And there is also the logic that we only want to withdraw when the fee is below a certain percentage.
This is logic that is specific to the Kraken marketplace, because it has a fixed fee. If the fee scheme is different for another marketplace, we would want to have different logic. So it is okay to put it into the Kraken implementation. It is not okay that the code cannot be tested, though.
Adding a new seam
I've asked on Stack Exchange to find out whether there are better ways to do it. And, unsurprising in retrospect, the answer has been to split the code one more time. I've already separated the concrete marketplace implementation from the remainder of the code with an interface (dependency inversion), but I need to do it again, just a bit different.
What I need to do is write a mock for the krakenex
package. Doing the depedency inversion again would introduce an interface for a Kraken library and a concrete implementation which is an adapter. But for real dependency inversion that interface would have to be under the control of the business code, not at the control of the external dependency. Therefore I would model it according to my needs, not model it according to the external package. I would end up with exactly the situation that I have right now. So this is not the way to go. Instead I implement a mock variant of the package such that I can mimic the Kraken API. This will not be a perfect replacement that tests everything, but I can then separate out the code which parses the results from the krakenex
package. And then I can test that.
The design will change just slightly. I introduce an interface which is exactly what the krakenex
provides. Therefore this interface is not under my control, but it is controlled by the external library. This is no dependency inversion. Still I am able to insert either the original module or my mock, which I will have to write.
This has added another layer in the system. In the language of Michael Feathers1 this is a “seam” where one can cut it apart and sew together in a different way.
Implementing the mock
In order to implement the mock, I need to know what the API actually returns. The krakenex
package is just a little wrapper around the Kraken API, it doesn't really change the names or structure of the response. It just handles the cryptography. This means that the official documentation of the REST API contains what I want.
Let us start with the spot price. This is one of the public methods. At the moment my implementation looks like this:
def get_spot_price(self, coin: str, fiat: str, now: datetime.datetime) -> Price: answer = self.handle.query_public( "Ticker", {"pair": f"{map_normal_to_kraken(coin)}{fiat}"} ) raise_error(answer, TickerError) close = float(list(answer["result"].values())[0]["c"][0]) logger.debug(f"Retrieved {close} for {fiat}/{coin} from Krakenex.") price = Price(timestamp=now, last=close, coin=coin, fiat=fiat) return price
We will need to implement this method. The official documentation for Ticker
gives this example:
$ curl "https://api.kraken.com/0/public/Ticker?pair=XBTUSD" {"error":[],"result":{"XXBTZUSD":{"a":["56576.30000","9","9.000"],"b":["56576.20000","1","1.000"],"c":["56576.30000","0.00006557"],"v":["735.69529661","2051.24626180"],"p":["56549.17405","56591.62229"],"t":[6915,27793],"l":["56127.80000","56058.20000"],"h":["57063.80000","57410.20000"],"o":"56528.20000"}}}
We get a JSON response. It always has an error
key, and a result
key. Then the result has one entry per currency pair requested. For each pair we have various prices. I am only interested in the close price, denoted by "c"
. The first entry is the price, the second one is the volume of the last trade.
This explains the strange answer["result"].values())[0]["c"][0]
that I have in the get_spot_price()
method. Honestly I forgot why exactly I had this, and only by reading the API documentation I understand again.
If we pass a currency pair which isn't supported, we get an error string in the result.
$ curl "https://api.kraken.com/0/public/Ticker?pair=XXXYYY" {"error":["EQuery:Unknown asset pair"]}
Implementing this seems to be easy. I create a new class, and give it the same methods that the krakenex.API
class has. I haven't formally defined the interface yet, but using duck typing in Python it is not an issue. Only the Ticker
operation is implemented yet, and it will always give the price it had at the time I ran the curl
command.
class KrakenexMock: def query_public(self, command: str, parameters: Dict) -> Dict: if command == "Ticker": return self._ticker(parameters) else: raise NotImplementedError() def _ticker(self, parameters: Dict) -> Dict: if parameters["pair"] in ["XBTEUR", "BTCEUR", "XXBTZEUR"]: return { "error": [], "result": { "XXBTZEUR": { "a": ["50162.20000", "1", "1.000"], "b": ["50162.10000", "2", "2.000"], "c": ["50162.20000", "0.00196431"], "v": ["1194.93544125", "3142.87839034"], "p": ["50218.07897", "50141.26546"], "t": [7355, 32353], "l": ["49750.00000", "49517.80000"], "h": ["50552.70000", "50657.00000"], "o": "50023.50000", } }, } else: return {"error": ["EQuery:Unknown asset pair"]}
Using the mock
Now that we have the mock for a single operation, I need to make sure that the KrakenexMarketplace
class can be instantiated with the mock. For now I will just pass a special KrakenConfig.key
value to select the mock.
class KrakenexMarketplace(Marketplace): def __init__(self, config: KrakenConfig): if config.key == 'mock': self.handle = KrakenexMock() else: self.handle = krakenex.API(config.key, config.secret) self.withdrawal_config = config.withdrawal self.prefer_fee_in_base_currency = config.prefer_fee_in_base_currency
And with that I was then able to write a test for the get_spot_price()
method:
def test_get_spot_price_success() -> None: config = KrakenConfig('mock', 'mock', False, {}) market = KrakenexMarketplace(config) now = datetime.datetime.now() price = market.get_spot_price("BTC", "EUR", now) assert price.timestamp == now assert price.coin == 'BTC' assert price.fiat == "EUR" assert price.last == 50162.2
This tests exercises the mapping from “normal” currency names to Kraken's strange “X” and “Z” prefixes. It also exercises the code which cuts out the value that we want from the answer.
I also need to ensure that the TickerError
is raised properly when an illegal currency pair is passed. This is a rather easy test to write.
def test_get_spot_price_error() -> None: config = KrakenConfig("mock", "mock", False, {}) market = KrakenexMarketplace(config) now = datetime.datetime.now() with pytest.raises(TickerError): price = market.get_spot_price("AAA", "AAA", now)
The coverage of the krakenex_adaptor.py
module was at 28 % before the changes. Now it is at 59 %, which is a great improvement. The get_spot_price()
method now has full coverage.
There are more methods to change like this. Now that I can instantiate the KrakenexMarketplace
, I can also test the get_name()
function. This brought the coverage up to 60 %.
Testing the Balance operation
For the Ticker
operation it was okay to implement fixed answers into the mock object. But the next one that I want to do is the Balance
operation. It doesn't have any parameters, and it can have three results:
- The user has funds on the marketplace, a dictionary with all the balances is returned.
- The user has no funds, then not even the
results
dictionary is returned. - Some error occured, there is a string in the
errors
list.
Because the API call has no parameters, I need to somehow inject that state externally. I could embellish my mock object for that, but likely this is already a solved problem. So I will take a look at the PyTest documentation for monkeypatching. There one can patch existing objects, which I could use to mock the krakenex
library. But it doesn't provide me with an easy fake object.
I will change the code such that the KrakenexMarketplace
will get the implementation of the KrakenexInterface
passed during construction. This will complicate the factory function for the default case, but it gives me complete control over it in the tests.
For this I will declare the interface. Because I cannot make the krakenex.API
class derive from my class, I need to use a protocol to define the specification.
class KrakenexInterface(Protocol): def query_public(self, command: str, parameters: Dict = None) -> Dict: raise NotImplementedError() def query_private(self, command: str, parameters: Dict = None) -> Dict: raise NotImplementedError()
Then I have implemented a simple class which lets me specify arbitrary functions to handle the different API commands:
class KrakenexMock: def __init__(self, methods: Dict[str, Callable] = None): if methods: self.methods = methods else: self.methods = {} def query_public(self, command: str, parameters: Dict = None) -> Dict: return self.methods[command](parameters) def query_private(self, command: str, parameters: Dict = None) -> Dict: return self.methods[command](parameters)
In the KrakenexMarketplace
I just let the client pass such a handle to the API, otherwise it will construct the real thing.
class KrakenexMarketplace(Marketplace): def __init__(self, config: KrakenConfig, handle: KrakenexInterface = None): if handle: self.handle = handle else: self.handle = krakenex.API(config.key, config.secret) self.withdrawal_config = config.withdrawal self.prefer_fee_in_base_currency = config.prefer_fee_in_base_currency
This lets me rewrite the test for the spot price with the expected value right there. I have also removed all the test code from the actual library, the mock has moved completely into the test code.
def test_get_spot_price_error() -> None: def ticker(parameters: Dict) -> Dict: return {"error": ["EQuery:Unknown asset pair"]} krakenex_interface = KrakenexMock({"Ticker": ticker}) config = KrakenConfig("mock", "mock", False, {}) market = KrakenexMarketplace(config, krakenex_interface) now = datetime.datetime.now() with pytest.raises(TickerError): price = market.get_spot_price("AAA", "AAA", now)
And with that I can write the tests for the balance that I have in mind. First comes the test for the filled wallet.
def test_balance_full() -> None: def mock_balance(parameters: Dict) -> Dict: return { "error": [], "result": { "ZEUR": "6789.1234", "XXBT": "1234.5678", }, } krakenex_interface = KrakenexMock({"Balance": mock_balance}) config = KrakenConfig("mock", "mock", False, {}) market = KrakenexMarketplace(config, krakenex_interface) now = datetime.datetime.now() balances = market.get_balance() assert balances == {"EUR": 6789.1234, "BTC": 1234.5678}
And then I can write one for the empty wallet and the error case. These just take parts which we had used before, so I will not paste the code here. The coverage has gone up to 68 % now.
Conclusion
Following the same steps for the remainder of the module, I was able to get the coverage to a full 100 %. Now all the cases are covered, all error conditions from the API are tested. This means that I can have even more confidence that the code is working as intended. And if a bug is reported in the future, I can write a regression test with the exact answer of the API.
-
Michael Feathers. Working Effectively with Legacy Code (2004) ↩