diff --git a/README.md b/README.md deleted file mode 100644 index 8236bd9..0000000 --- a/README.md +++ /dev/null @@ -1,3 +0,0 @@ -# Test Driven Python Development - -This repository contains the code for the book Test Driven Python Development (Packt Publishing). The repository has branches, so you can get the code for the individual chapters by choossing the appropriate branch. Feel free to fork the repository to work on the code as you go through the book. diff --git a/README.rst b/README.rst new file mode 100644 index 0000000..d3f83e6 --- /dev/null +++ b/README.rst @@ -0,0 +1,24 @@ +Test Driven Python Development +============================== + +This repository contains the code for the book `Test Driven Python Development `. + +How to use this repository +-------------------------- + +The repository has tags, so you can get the code for the individual chapters by choosing the appropriate tag. + +**If you are familiar with using git & github**: Clone this repository, then look at tags to get the point in the code you want to go to. Checkout the appropriate tag on your local repository to get to that particular state of the code. + +**If you have not used github before**: Look at the `list of tags `, and click the *zip link* for the point in the code that you are interested in. This will download the code for that particular commit as a zip file. Unzip this file into your project folder and work from there. + +Notes +----- + +* Python 2.6+ compatible version of the code is available on it's own branch here - https://github.com/siddhi/test_driven_python/tree/py2.6 +* Some readers have submitted contributions to this repository. You can access the code with these contributions in the *contrib* branch - https://github.com/siddhi/test_driven_python/tree/contrib + +Submitting contributions +------------------------ + +If you would like to submit changes to this repository, checkout the *contrib* branch, make changes and submit a pull request. I will be happy to merge the changes in. diff --git a/nose2.cfg b/nose2.cfg new file mode 100644 index 0000000..9988d74 --- /dev/null +++ b/nose2.cfg @@ -0,0 +1,19 @@ +[unittest] +test-file-pattern=test_*.py +test-method-prefix=test +plugins = nose2.plugins.coverage + nose2.plugins.junitxml + nose2.plugins.layers +exclude-plugins = nose2.plugins.doctest + +[layer-reporter] +always-on = True +colors = False + +[junit-xml] +always-on = False +path = nose2.xml + +[coverage] +always-on = False +coverage-report = ["html", "xml"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1f671a5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +nose2 \ No newline at end of file diff --git a/run_test.py b/run_test.py new file mode 100644 index 0000000..5595d1a --- /dev/null +++ b/run_test.py @@ -0,0 +1,24 @@ +import unittest + + +class AttribLoader(unittest.TestLoader): + def __init__(self, attrib): + self.attrib = attrib + + def loadTestsFromModule(self, module, use_load_tests=False): + return super().loadTestsFromModule(module, use_load_tests=False) + + def getTestCaseNames(self, testCaseClass): + test_names = super().getTestCaseNames(testCaseClass) + filtered_test_names = [test + for test in test_names + if hasattr(getattr(testCaseClass, test), + self.attrib)] + return filtered_test_names + + +if __name__ == "__main__": + loader = AttribLoader("slow") + test_suite = loader.discover(".") + runner = unittest.TextTestRunner() + runner.run(test_suite) diff --git a/stock_alerter/processor.py b/stock_alerter/processor.py new file mode 100644 index 0000000..6077fda --- /dev/null +++ b/stock_alerter/processor.py @@ -0,0 +1,9 @@ +class Processor: + def __init__(self, reader, exchange): + self.reader = reader + self.exchange = exchange + + def process(self): + for symbol, timestamp, price in self.reader.get_updates(): + stock = self.exchange[symbol] + stock.update(timestamp, price) diff --git a/stock_alerter/reader.py b/stock_alerter/reader.py new file mode 100644 index 0000000..8ec792f --- /dev/null +++ b/stock_alerter/reader.py @@ -0,0 +1,28 @@ +from datetime import datetime + + +class ListReader: + """Reads a series of updates from a list""" + def __init__(self, updates): + self.updates = updates + + def get_updates(self): + for update in self.updates: + yield update + + +class FileReader: + """Reads a series of stock updates from a file""" + def __init__(self, filename): + self.filename = filename + + def get_updates(self): + """Returns the next update everytime the method is called""" + with open(self.filename, "r") as fp: + data = fp.read() + lines = data.split() + for line in lines: + symbol, timestamp, price = line.split(",") + yield (symbol, + datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f"), + int(price)) diff --git a/stock_alerter/readme.txt b/stock_alerter/readme.txt new file mode 100644 index 0000000..a6cd892 --- /dev/null +++ b/stock_alerter/readme.txt @@ -0,0 +1,46 @@ +The stock_alerter module allows you to setup rules and get alerted when +those rules are met. + +>>> from datetime import datetime + +First, we need to setup an exchange which contains all the stocks that +are going to be processed. A simple dictionary will do. + +>>> from stock_alerter.stock import Stock +>>> exchange = {"GOOG": Stock("GOOG"), "AAPL": Stock("AAPL")} +>>> for key in sorted(exchange.keys()): +... print(key, exchange[key]) +... +AAPL +GOOG + +Next, we configure the reader. The reader is the source from where the +stock updates are coming. The module provides two readers out of the +box: A FileReader for reading updates from a comma separated file, +and a ListReader to get updates from a list. You can create other +readers, such as an HTTPReader to get updates from a remote server. +Here we create a simple ListReader by passing in a list of 3-tuples +containing the stock symbol, timestamp and price. + +>>> from stock_alerter.reader import ListReader +>>> reader = ListReader([("GOOG", datetime(2014, 2, 8), 5)]) + +Next, we setup an Alert. We give it a rule, and an action to be taken +when the rule is fired. + +>>> from stock_alerter.alert import Alert +>>> from stock_alerter.rule import PriceRule +>>> from stock_alerter.action import PrintAction +>>> alert = Alert("GOOG > $3", PriceRule("GOOG", lambda s: s.price > 3),\ +... PrintAction()) + +Connect the alert to the exchange + +>>> alert.connect(exchange) + +Now that everything is setup, we can start processing the updates + +>>> from stock_alerter.processor import Processor +>>> processor = Processor(reader, exchange) +>>> processor.process() +GOOG > $3 diff --git a/stock_alerter/stock.py b/stock_alerter/stock.py index e8bcddb..49893af 100644 --- a/stock_alerter/stock.py +++ b/stock_alerter/stock.py @@ -22,19 +22,70 @@ def __init__(self, symbol): @property def price(self): + """Returns the current price of the Stock + + >>> stock.update(datetime(2011, 10, 3), 10) + >>> stock.price + 10 + + The method will return the latest price by timestamp, so even if + updates are out of order, it will return the latest one + + >>> stock = Stock("GOOG") + >>> stock.update(datetime(2011, 10, 3), 10) + + Now, let us do an update with a date that is earlier than the previous + one + + >>> stock.update(datetime(2011, 10, 2), 5) + + And the method still returns the latest price + + >>> stock.price + 10 + + If there are no updates, then the method returns None + + >>> stock = Stock("GOOG") + >>> print(stock.price) + None + """ try: return self.history[-1].value except IndexError: return None def update(self, timestamp, price): + """Updates the stock with the price at the given timestamp + + >>> stock.update(datetime(2014, 10, 2), 10) + >>> stock.price + 10 + + The method raises a ValueError exception if the price is negative + + >>> stock.update(datetime(2014, 10, 2), -1) + Traceback (most recent call last): + ... + ValueError: price should not be negative + """ if price < 0: raise ValueError("price should not be negative") self.history.update(timestamp, price) self.updated.fire(self) def is_increasing_trend(self): - return self.history[-3].value < self.history[-2].value < self.history[-1].value + """Returns True if the past three values have been strictly increasing + + Returns False if there have been less than three updates so far + + >>> stock.is_increasing_trend() + False + """ + try: + return self.history[-3].value < self.history[-2].value < self.history[-1].value + except IndexError: + return False def _is_crossover_below_to_above(self, on_date, ma, reference_ma): prev_date = on_date - timedelta(1) diff --git a/stock_alerter/tests/test_alert.py b/stock_alerter/tests/test_alert.py index d7f38e1..dc83efe 100644 --- a/stock_alerter/tests/test_alert.py +++ b/stock_alerter/tests/test_alert.py @@ -22,3 +22,31 @@ def test_action_is_executed_when_rule_matches(self): alert.connect(exchange) exchange["GOOG"].update(datetime(2014, 2, 10), 11) action.execute.assert_called_with("sample alert") + + def test_action_doesnt_fire_if_rule_doesnt_match(self): + goog = Stock("GOOG") + exchange = {"GOOG": goog} + rule = PriceRule("GOOG", lambda stock: stock.price > 10) + rule_spy = mock.MagicMock(wraps=rule) + action = mock.MagicMock() + alert = Alert("sample alert", rule_spy, action) + alert.connect(exchange) + alert.check_rule(goog) + rule_spy.matches.assert_called_with(exchange) + self.assertFalse(action.execute.called) + + def test_action_fires_when_rule_matches(self): + goog = Stock("GOOG") + exchange = {"GOOG": goog} + main_mock = mock.MagicMock() + rule = main_mock.rule + rule.matches.return_value = True + rule.depends_on.return_value = {"GOOG"} + action = main_mock.action + alert = Alert("sample alert", rule, action) + alert.connect(exchange) + goog.update(datetime(2014, 5, 14), 11) + self.assertEqual([mock.call.rule.depends_on(), + mock.call.rule.matches(exchange), + mock.call.action.execute("sample alert")], + main_mock.mock_calls) diff --git a/stock_alerter/tests/test_doctest.py b/stock_alerter/tests/test_doctest.py new file mode 100644 index 0000000..e5da804 --- /dev/null +++ b/stock_alerter/tests/test_doctest.py @@ -0,0 +1,19 @@ +import doctest +from datetime import datetime + +from stock_alerter import stock + + +def setup_stock_doctest(doctest): + s = stock.Stock("GOOG") + doctest.globs.update({"stock": s}) + + +def load_tests(loader, tests, pattern): + tests.addTests(doctest.DocTestSuite(stock, globs={ + "datetime": datetime, + "Stock": stock.Stock + }, setUp=setup_stock_doctest)) + options = doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE + tests.addTests(doctest.DocFileSuite("readme.txt", package="stock_alerter", optionflags=options)) + return tests diff --git a/stock_alerter/tests/test_reader.py b/stock_alerter/tests/test_reader.py new file mode 100644 index 0000000..d6f76c5 --- /dev/null +++ b/stock_alerter/tests/test_reader.py @@ -0,0 +1,18 @@ +import unittest +from unittest import mock +from datetime import datetime + +from ..reader import FileReader + + +class FileReaderTest(unittest.TestCase): + @mock.patch("builtins.open", + mock.mock_open(read_data="""\ +GOOG,2014-02-11T14:10:22.13,10""")) + def test_FileReader_returns_the_file_contents(self): + reader = FileReader("stocks.txt") + updater = reader.get_updates() + update = next(updater) + self.assertEqual(("GOOG", + datetime(2014, 2, 11, 14, 10, 22, 130000), + 10), update) diff --git a/stock_alerter/tests/test_stock.py b/stock_alerter/tests/test_stock.py index a7453d4..cfb644b 100644 --- a/stock_alerter/tests/test_stock.py +++ b/stock_alerter/tests/test_stock.py @@ -36,26 +36,24 @@ def test_price_is_the_latest_even_if_updates_are_made_out_of_order(self): class StockTrendTest(unittest.TestCase): - def setUp(self): - self.goog = Stock("GOOG") - - def given_a_series_of_prices(self, prices): + def given_a_series_of_prices(self, goog, prices): timestamps = [datetime(2014, 2, 10), datetime(2014, 2, 11), datetime(2014, 2, 12), datetime(2014, 2, 13)] for timestamp, price in zip(timestamps, prices): - self.goog.update(timestamp, price) - - def test_increasing_trend_is_true_if_price_increase_for_3_updates(self): - self.given_a_series_of_prices([8, 10, 12]) - self.assertTrue(self.goog.is_increasing_trend()) - - def test_increasing_trend_is_false_if_price_decreases(self): - self.given_a_series_of_prices([8, 12, 10]) - self.assertFalse(self.goog.is_increasing_trend()) - - def test_increasing_trend_is_false_if_price_equal(self): - self.given_a_series_of_prices([8, 10, 10]) - self.assertFalse(self.goog.is_increasing_trend()) + goog.update(timestamp, price) + + def test_stock_trends(self): + dataset = [ + ([8, 10, 12], True), + ([8, 12, 10], False), + ([8, 10, 10], False) + ] + for data in dataset: + prices, output = data + with self.subTest(prices=prices, output=output): + goog = Stock("GOOG") + self.given_a_series_of_prices(goog, prices) + self.assertEqual(output, goog.is_increasing_trend()) class StockCrossOverSignalTest(unittest.TestCase):