diff --git a/lib/reports.py b/lib/reports.py index b630c72b..b875fa60 100755 --- a/lib/reports.py +++ b/lib/reports.py @@ -49,7 +49,7 @@ def prepare_watchlist(service, user, action=None, ticker=None): ticker_link = util.finance_link(ticker, ticker, service) duplicate = False transformed = False - watchlist = util.json_load('finbot_watchlist.json', persist=True) + watchlist = util.json_load('finbot_watchlist.json', persist=True) or [] # absent until the first add if action == 'add': if ticker in watchlist: duplicate = True @@ -301,6 +301,7 @@ def last_col(e): link = util.finance_link(ticker, market_data[ticker]['profile_exchange'], service) flag = util.flag_from_ticker(ticker) payload_staging.append(f"{flag} {title} ({link}) mkt cap: {market_cap_readable} {market_cap}") + payload = ['No market cap data found'] if payload_staging: payload_staging.sort(key=last_col) if action == 'top': diff --git a/lib/util.py b/lib/util.py index 03c09a39..d6ebacbd 100644 --- a/lib/util.py +++ b/lib/util.py @@ -749,6 +749,14 @@ def hbar_chart(rows, title, subtitle='', value_fmt=None, axis_fmt=None, threshol canvas.draw() label_width = [t.get_window_extent(renderer).width for t in base_labels] value_width = [t.get_window_extent(renderer).width for t in value_labels] + if xlim: # zoomed charts put the labels in the left margin: shrink the text until the widest label fits inside the image + room = left * fig.get_figwidth() * dpi - 30 # also leaves the 4pt gap before the bar base + if max(label_width) > room: + for t in base_labels + value_labels: + t.set_fontsize(t.get_fontsize() * room / max(label_width)) + canvas.draw() + label_width = [t.get_window_extent(renderer).width for t in base_labels] + value_width = [t.get_window_extent(renderer).width for t in value_labels] width = ax.get_window_extent(renderer).width lo, hi = lo0, hi0 inside = [True] * len(real) @@ -828,7 +836,7 @@ def days_from_human_days(arg): def get_holdings_and_watchlist(): tickers = set(sharesight.get_holdings_wrapper()) - tickers.update(json_load('finbot_watchlist.json', persist=True)) + tickers.update(json_load('finbot_watchlist.json', persist=True) or []) # absent until the first add if 'GOOG' in tickers and 'GOOGL' in tickers: tickers.remove("GOOGL") tickers = sorted(set(tickers)) diff --git a/lib/worker.py b/lib/worker.py index 1dbba528..bc8c7e14 100755 --- a/lib/worker.py +++ b/lib/worker.py @@ -270,7 +270,7 @@ def alliterate(): typing.stop() elif m_performance: portfolio_select = None - days = config_past_days + days = config_past_days or 7 # performance needs a period; past_days = 0 means "today" elsewhere for arg in m_performance.groups()[:2]: # groups 2 and 3, allow arbitrary order if arg: try: @@ -330,7 +330,7 @@ def alliterate(): price_percent = float(arg.split('%')[0]) except ValueError: if arg == m_price.group(1): - if arg == 'top': + if arg in ('top', 'bottom'): # one chart shows both ends price_percent=0 top=10 else: diff --git a/lib/yahoo.py b/lib/yahoo.py index 562b9bbe..76450e15 100644 --- a/lib/yahoo.py +++ b/lib/yahoo.py @@ -104,7 +104,7 @@ def fetch(tickers): # DO NOT CACHE MORE THAN 5 mins if not len(tickers): print("no tickers provided to yahoo.fetch()", file=sys.stderr) - return None + return {} tickers = sorted(set(tickers)) # de-dupe tickers_sha256 = hashlib.sha256(str.encode("_".join(tickers))).hexdigest() if config_cache: @@ -136,7 +136,7 @@ def fetch(tickers): data = data['quoteResponse'] if data['result'] is None or data['error'] is not None: print(f"{tickers}†", sep=' ', end='', flush=True, file=sys.stderr) - return None + return {} # callers iterate the result for item in data['result']: ticker = item['symbol'] try: @@ -292,7 +292,7 @@ def fetch_detail(ticker, seconds=config_cache_seconds): # upgradeDowngradeHistory try: profile_title = data['quoteSummary']['result'][0]['price']['longName'] - except (KeyError, IndexError, ValueError): + except (KeyError, IndexError, ValueError, TypeError): # TypeError: Yahoo answers result=null for unknown symbols print(ticker + 'x', sep=' ', end='', flush=True, file=sys.stderr) return {} if profile_title is None: diff --git a/price.py b/price.py index 6c0b4995..abccbe2a 100755 --- a/price.py +++ b/price.py @@ -166,7 +166,7 @@ def prepare_price_payload(service, market_data, threshold, _rows_out=None): chart_top = chart_rows[:top] chart_bottom = [row for row in chart_rows[-top:] if row not in chart_top] payload = payload[:top] - heading = f'Top {top} performers {util.date_range_english(days, as_of)}' + heading = f'Top {top} performers {util.date_range_english(days, as_of) if days else on}' elif days: heading = f'Price movement {threshold}%, {util.date_range_english(days, as_of)}' else: diff --git a/rating.py b/rating.py index 14774f61..b012c627 100755 --- a/rating.py +++ b/rating.py @@ -49,9 +49,9 @@ def get_emoji(old, new): else: if interactive: if specific_stock: - payload = [f"{emoji}No rating changes found for {tickers[0]}"] + payload = [f"No rating changes found for {tickers[0]}"] else: - payload = [f"{emoji}No rating changes found"] + payload = ["No rating changes found"] return payload, new # MAIN # diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/common.py b/tests/common.py new file mode 100644 index 00000000..6217fc1b --- /dev/null +++ b/tests/common.py @@ -0,0 +1,87 @@ +"""Shared fixture: points finbot at a temp cache dir and replaces every external service +(Telegram, Yahoo, Sharesight, Short Man) with canned data. Nothing here touches the network.""" +import io, os, sys, tempfile, time, unittest +import pandas as pd +from unittest import mock + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) +os.chdir(ROOT) +os.environ.update(telegramOutgoingWebhook='https://example.com/telegram', telegramOutgoingToken='tok') + +from lib import webhook, util, yahoo, sharesight, shortman, worker, charts, telegram, reports +import performance, shorts, price, reminder, cal, trades, milestone, rating +import bot # imported once up front: it monkey-patches with gevent, which must not happen mid-test + +REAL = {(m.__name__, n): getattr(m, n) for m, n in ((yahoo, 'fetch'), (yahoo, 'fetch_detail'), (yahoo, 'price_series'), + (sharesight, 'get_portfolios'), (sharesight, 'get_trades'), (sharesight, 'get_performance'), (sharesight, 'get_performance_wrapper'))} +REAL_GET_HOLDINGS = util.get_holdings_and_watchlist +REAL_PAYLOAD_WRAPPER = webhook.payload_wrapper # FinbotCase replaces it; tests of the real thing use this +TICKERS = [f'T{i}' for i in range(25)] +BOT = '@bot' + + +STATES = ('REGULAR', 'PRE', 'POST', 'CLOSED') + + +def market_entry(ticker): + i = int(ticker[1:]) if ticker[1:].isdigit() else 0 # varied data so threshold/sort/sell branches all have rows + return {'marketState': STATES[i % 4], 'regularMarketTime': time.time(), 'quoteType': 'EQUITY', + 'profile_title': 'Co ' + ticker, 'profile_exchange': 'NasdaqGS', 'percent_change': (i - 12) * 1.5, 'percent_change_premarket': (i - 12) * 1.5, 'percent_change_postmarket': (i - 12) * 1.5, + 'market_cap': 1e9, 'price_to_earnings_trailing': 9, 'price_to_earnings_forward': 9, + 'beta': 2.0, 'short_percent': 20, 'fiftyTwoWeekHigh': 120.0, 'fiftyTwoWeekLow': 80.0, + 'regularMarketPrice': 100.0, 'earnings_date': time.time() + 86400 * 3, + 'ex_dividend_date': time.time() + 86400 * 3, 'recommend': 'buy' if i % 2 else 'sell', 'recommend_index': 1.2 if i % 2 else 4.2, + 'recommend_analysts': 4, 'price_to_earnings_peg': 1.2, 'currency': 'USD'} + + +class FinbotCase(unittest.TestCase): + """Base class: self.sent collects ('photo'|'text', ...) tuples for everything the bot tried to send.""" + + def patch(self, target, attr, value): + p = mock.patch.object(target, attr, value) + p.start() + self.addCleanup(p.stop) + + def setUp(self): + fake = mock.Mock(status_code=200, text='{}') + fake.json.return_value = {'ok': True, 'result': {'message_id': 1, 'chat': {'id': 5}}} + self.patch(__import__('requests').Session, 'request', mock.Mock(return_value=fake)) # no real network traffic + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.patch(util, 'config_var_dir', tmp.name) + self.patch(util, 'config_cache_dir', tmp.name) + self.patch(webhook, 'webhooks', {'telegram': 'https://api.telegram.org/botX/'}) + for m in (performance, shorts, price, cal, trades, milestone, rating, worker, reminder): + self.patch(m, 'webhooks', webhook.webhooks) + self.sent = [] + self.errors = [] + self.patch(webhook, 'report_error', lambda e, *a, **k: self.errors.append(webhook.error_line(e, k.get('context')))) + self.patch(webhook, 'sendPhoto', lambda chat, img, cap, svc, **k: self.sent.append(('photo', len(img.read()), cap.split('\n')[0]))) + self.patch(webhook, 'payload_wrapper', lambda svc, url, payload, *a, **k: self.sent.append(('text', payload)) or []) + self.patch(util, 'get_holdings_and_watchlist', lambda: TICKERS) + self.patch(yahoo, 'fetch', lambda ts: {t: market_entry(t) for t in ts}) + self.patch(yahoo, 'fetch_detail', lambda t, *a, **k: {t: market_entry(t)}) + self.patch(yahoo, 'price_history', lambda t, days=None, **k: ({days: 1.5} if days else {'Max': 5, '1Y': 3}, None)) + self.patch(yahoo, 'prefetch_history', lambda ts: None) + self.patch(yahoo, 'historic_high', lambda t: (100.0, 50.0)) + dates = pd.date_range(end=pd.Timestamp.today().normalize(), periods=400) + self.patch(yahoo, 'price_series', lambda t: ('Co ' + t, 'NASDAQ', pd.Series(range(100, 500), index=dates, name=t, dtype=float))) + self.patch(shortman, 'fetch', lambda md: md) + holdings = [{'id': i, 'instrument': {'code': t, 'market_code': 'NASDAQ'}, 'portfolio': {'name': 'Rob'}, + 'capital_gain_percent': str(i * 4 - 40), 'value': '1000'} for i, t in enumerate(TICKERS)] + self.patch(sharesight, 'get_performance_wrapper', lambda d: {1: {'report': {'holdings': holdings}}}) + self.patch(sharesight, 'get_portfolios', lambda: {'Rob': 1}) + self.patch(sharesight, 'get_performance', lambda pid, d, *a: {'report': { + 'holdings': holdings, 'currency_gain_percent': '0', 'capital_gain_percent': '2.5', 'total_gain_percent': '0'}}) + self.patch(sharesight, 'get_trades', lambda name, pid, days=7: [{ + 'id': 1, 'transaction_type': 'BUY', 'symbol': 'T1', 'market': 'NASDAQ', 'portfolio': name, + 'transaction_date': time.strftime('%Y-%m-%d'), 'quantity': '10', 'price': '5.0', 'value': '50', 'brokerage_currency_code': 'USD', 'holding_id': 7}]) + + def command(self, text, chat='55'): + """Run a chat message through the real command parser and handlers.""" + self.sent.clear() + self.errors.clear() + worker.process_request('telegram', chat, '@u', text, BOT, 'U', '1') + self.assertEqual(self.errors, [], f'{text} crashed') # the worker reports crashes instead of raising + return list(self.sent) diff --git a/tests/golden/compare_lines.png b/tests/golden/compare_lines.png new file mode 100644 index 00000000..6f4f094f Binary files /dev/null and b/tests/golden/compare_lines.png differ diff --git a/tests/golden/hbar_mixed.png b/tests/golden/hbar_mixed.png new file mode 100644 index 00000000..2cf896ab Binary files /dev/null and b/tests/golden/hbar_mixed.png differ diff --git a/tests/golden/price_line.png b/tests/golden/price_line.png new file mode 100644 index 00000000..5992ce6c Binary files /dev/null and b/tests/golden/price_line.png differ diff --git a/tests/test_charts.py b/tests/test_charts.py new file mode 100644 index 00000000..6f02f465 --- /dev/null +++ b/tests/test_charts.py @@ -0,0 +1,242 @@ +"""Charts, tested three ways that survive restyling: + 1. image sanity: a real 1280x800 PNG, drawn, with nothing touching the edges (no clipped text) + 2. contracts: blue/red by sign, row order, titles and sort order, captured from what is handed to matplotlib + 3. one golden image per chart type, compared with a tolerance. Refresh after an intended restyle with: + UPDATE_GOLDEN=1 python -m unittest tests.test_charts +""" +import datetime, io, os, unittest +from unittest import mock +import numpy as np +import pandas as pd +from PIL import Image +from matplotlib.axes import Axes +from matplotlib.figure import Figure +from tests.common import * + +GOLDEN = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'golden') +BG, BLUE, RED = (0x14, 0x18, 0x1f), (0x39, 0x87, 0xe5), (0xfb, 0x71, 0x85) + +MIXED = [('AAA', 42), ('BBB', 17), ('CCC', 3), ('', None), ('DDD', -5), ('EEE', -30)] + + +def decode(buf): + buf.seek(0) + return Image.open(buf).convert('RGB') + + +def price_frame(days=120): + dates = [datetime.date(2026, 1, 1) + datetime.timedelta(days=i) for i in range(days)] + closes = [100 + 12 * np.sin(i / 9) + i * 0.15 for i in range(days)] + return pd.DataFrame({'Date': dates, 'Close': closes}) + + +def compare_series(): + index = pd.date_range('2026-01-01', periods=120) + return [(name, pd.Series(100 + np.cumsum(np.sin(np.arange(120) / k) + drift), index=index)) + for name, k, drift in (('AAA', 7, 0.2), ('BBB', 11, -0.1), ('CCC', 5, 0.05))] + + +CHARTS = { + 'hbar_mixed': lambda: util.hbar_chart(MIXED, 'Price movement 5.0%, 5 Sep - 5 Oct 2026', 'subtitle here'), + 'price_line': lambda: util.graph(price_frame(), 'Apple Inc (AAPL)', 'USD'), + 'compare_lines': lambda: util.compare_graph(compare_series(), 'Compare 120 days', 'subtitle here'), +} +# inputs that have broken chart code before or are likely to: sizes, signs, long names, zeros, a single row +EDGE_ROWS = { + 'one_row': [('AAA', 5)], + 'one_negative': [('AAA', -5)], + 'all_positive': [(f'T{i}', i + 1) for i in range(15)], + 'all_negative': [(f'T{i}', -(i + 1)) for i in range(15)], + 'all_zero': [(f'T{i}', 0) for i in range(5)], + 'forty_rows': [(f'T{i}', (i - 20) * 1.5) for i in range(40)], + 'long_labels': [('A very long company name indeed (TICKER)', 12), ('Another equally long label here ok', -8)], + 'huge_values': [('AAA', 12345), ('BBB', -9876)], + 'tiny_values': [('AAA', 0.01), ('BBB', -0.02)], +} + + +class ImageSanity(unittest.TestCase): + def check_image(self, buf, name): + image = decode(buf) + self.assertEqual(image.size, (1280, 800), name) + pixels = np.asarray(image) + self.assertGreater(len({tuple(p) for p in pixels.reshape(-1, 3)[::97]}), 8, f'{name} looks blank') + self.assertGreater(pixels.std(), 5, f'{name} looks blank') + np.testing.assert_array_equal(pixels[0, 0], BG, err_msg=name) + edge = np.concatenate([pixels[:3].reshape(-1, 3), pixels[-3:].reshape(-1, 3), pixels[:, :3].reshape(-1, 3), pixels[:, -3:].reshape(-1, 3)]) + self.assertTrue((edge == BG).all(), f'{name}: something is drawn on the image edge (clipped text or bars?)') + + def test_each_chart_type(self): + for name, make in CHARTS.items(): + with self.subTest(chart=name): + self.check_image(make(), name) + + def test_edge_case_rows(self): + for name, rows in EDGE_ROWS.items(): + with self.subTest(rows=name): + self.check_image(util.hbar_chart(rows, 'Title', 'sub'), name) + self.check_image(util.rows_chart(rows, 'Title'), name + '/rows_chart') + + def test_zoomed_axis_and_threshold(self): + self.check_image(util.hbar_chart([('AAA', 1.2), ('BBB', 3.4)], 'Ratings', xlim=(0.9, 4.0)), 'zoomed') + self.check_image(util.hbar_chart([('WWWWWWWW', 1.2), ('MMM', 3.4)], 'Ratings', xlim=(0.9, 4.0)), 'zoomed wide labels') + self.check_image(util.hbar_chart([(f'T{i}', 1 + i / 10) for i in range(15)], 'Ratings', xlim=(0.9, 3.0)), 'zoomed 15 rows') + self.check_image(util.hbar_chart(MIXED, 'Threshold', threshold=10), 'threshold') + + def test_bar_graph_has_top_and_bottom(self): + self.check_image(util.bar_graph([('AAA', 9)], [('ZZZ', -9)], 'Top and bottom'), 'bar_graph') + + def test_flat_and_tiny_price_series(self): + flat = pd.DataFrame({'Date': [datetime.date(2026, 1, 1) + datetime.timedelta(days=i) for i in range(10)], 'Close': [5.0] * 10}) + self.check_image(util.graph(flat, 'Flat', 'USD'), 'flat price') + two = price_frame(2) + self.check_image(util.graph(two, 'Two points', 'USD'), 'two points') + + +class Contracts(unittest.TestCase): + def capture_bars(self, rows, **kwargs): + calls = [] + real = Axes.barh + def spy(ax, y, width, *a, **k): + calls.append({'y': list(y), 'width': list(width), 'color': list(k.get('color', []))}) + return real(ax, y, width, *a, **k) + with mock.patch.object(Axes, 'barh', spy): + util.hbar_chart(rows, 'T', **kwargs) + return calls[0] + + def capture_text(self, make): + texts = [] + real = Figure.text + def spy(fig, x, y, s, *a, **k): + texts.append(s) + return real(fig, x, y, s, *a, **k) + with mock.patch.object(Figure, 'text', spy): + make() + return texts + + def test_bars_are_blue_when_positive_and_red_when_negative(self): + bars = self.capture_bars([('A', 3), ('B', 0), ('C', -2)]) + self.assertEqual(bars['color'], ['#3987e5', '#3987e5', '#fb7185']) + + def test_first_row_is_on_top_and_spacers_are_skipped(self): + bars = self.capture_bars(MIXED) + self.assertEqual(bars['width'], [42, 17, 3, -5, -30]) + self.assertEqual(bars['y'], [0, 1, 2, 4, 5]) # row 3 is the blank spacer + calls = [] + real = Axes.set_ylim + with mock.patch.object(Axes, 'set_ylim', lambda ax, *a, **k: (calls.append(a), real(ax, *a, **k))[1]): + util.hbar_chart(MIXED, 'T') + self.assertTrue(any(len(a) == 2 and a[0] > a[1] for a in calls), 'y axis must be inverted so row 0 is drawn first (top)') + + def test_pixels_follow_the_sign_of_the_data(self): + def count(rows): + pixels = np.asarray(decode(util.hbar_chart(rows, 'T'))).reshape(-1, 3) + return (pixels == BLUE).all(axis=1).sum(), (pixels == RED).all(axis=1).sum() + blue, red = count([('A', 5), ('B', 3)]) + self.assertGreater(blue, 1000); self.assertEqual(red, 0) + blue, red = count([('A', -5), ('B', -3)]) + self.assertEqual(blue, 0); self.assertGreater(red, 1000) + blue, red = count([('A', 5), ('B', -3)]) + self.assertGreater(blue, 1000); self.assertGreater(red, 1000) + + def test_longer_bar_is_drawn_longer(self): + pixels = np.asarray(decode(util.hbar_chart([('A', 40), ('B', 10)], 'T'))) + blue_rows = np.where((pixels == BLUE).all(axis=2).any(axis=1))[0] + top_band, bottom_band = pixels[blue_rows[0]], pixels[blue_rows[-1]] + self.assertGreater((top_band == BLUE).all(axis=1).sum(), 2 * (bottom_band == BLUE).all(axis=1).sum()) + + def test_titles_are_drawn(self): + texts = self.capture_text(lambda: util.hbar_chart(MIXED, 'My Title', 'My subtitle')) + self.assertIn('My Title', texts) + self.assertIn('My subtitle', texts) + texts = self.capture_text(lambda: util.graph(price_frame(), 'Apple Inc (AAPL)', 'USD')) + self.assertIn('Apple Inc (AAPL)', texts) + self.assertTrue(any('over period' in t for t in texts)) + + def test_price_line_colour_follows_direction(self): + def line_colour(closes): + frame = pd.DataFrame({'Date': [datetime.date(2026, 1, 1) + datetime.timedelta(days=i) for i in range(len(closes))], 'Close': closes}) + texts = [] + real = Figure.text + with mock.patch.object(Figure, 'text', lambda fig, x, y, s, *a, **k: (texts.append((s, k.get('color'))), real(fig, x, y, s, *a, **k))[1]): + util.graph(frame, 'T', 'USD') + return next(color for s, color in texts if 'over period' in s) + self.assertEqual(line_colour([1, 2, 3]), '#34d399') # up: green + self.assertEqual(line_colour([3, 2, 1]), '#fb7185') # down: red + self.assertEqual(line_colour([2, 3, 2]), '#9aa4b2') # flat + + +class ReportChartInputs(FinbotCase): + """What the reports hand to the chart: sort order, titles and dates, taken from the real report code.""" + + def run_report(self, call): + captured = [] + real = util.hbar_chart + def spy(rows, title, subtitle='', **kw): + captured.append({'rows': list(rows), 'title': title, 'subtitle': subtitle}) + return real(rows, title, subtitle, **kw) + self.patch(util, 'hbar_chart', spy) + call() + self.assertTrue(captured, 'no chart was drawn') + return captured[0] + + def test_price_top_lists_best_first_then_worst(self): + chart = self.run_report(lambda: self.command('.price top 7d')) + values = [v for _, v in chart['rows'] if v is not None] + top, bottom = values[:10], values[10:] + self.assertEqual(top, sorted(top, reverse=True)) + self.assertEqual(bottom, sorted(bottom)) # the worst performer comes first, like the caption's second list + self.assertGreater(min(top), max(bottom)) + self.assertIn(None, [v for _, v in chart['rows']]) # spacer between the two lists + + def test_price_threshold_chart_is_sorted_and_titled_with_the_threshold(self): + chart = self.run_report(lambda: self.command('.price 5%')) + values = [v for _, v in chart['rows'] if v is not None] + self.assertEqual(values, sorted(values, reverse=True)) + self.assertIn('5', chart['title']) + + def test_price_over_a_period_has_a_dated_title(self): + chart = self.run_report(lambda: self.command('.price 7d')) + self.assertRegex(chart['title'], r'\d{1,2} \w{3} - \d{1,2} \w{3} \d{4}') # 28 Sep - 5 Oct 2026 + + def test_performance_is_sorted_by_percent(self): + chart = self.run_report(lambda: self.command('.performance 7d')) + values = [v for _, v in chart['rows'] if v is not None] + self.assertEqual(values, sorted(values, reverse=True)) + + def test_shorts_is_sorted_and_titled(self): + chart = self.run_report(lambda: self.command('.shorts')) + values = [v for _, v in chart['rows'] if v is not None] + self.assertEqual(values, sorted(values, reverse=True)) + self.assertTrue(chart['title']) + + +class GoldenImages(unittest.TestCase): + """One reference image per chart type. Tolerant on purpose: compares 160x100 thumbnails, so font anti-aliasing and + minor matplotlib differences pass, while changed colours, layout, bar sizes or missing text do not.""" + TOLERANCE = 2.0 # mean absolute difference, in 0-255 grey levels, of the 160x100 thumbnails + + @staticmethod + def thumb(image): + return image.resize((160, 100), Image.LANCZOS) # goldens are stored at this size too: a few KB each, not 100 KB + + def check(self, name): + thumb = self.thumb(decode(CHARTS[name]())) + path = os.path.join(GOLDEN, name + '.png') + if os.environ.get('UPDATE_GOLDEN') or not os.path.exists(path): + os.makedirs(GOLDEN, exist_ok=True) + thumb.save(path, optimize=True) + if not os.environ.get('UPDATE_GOLDEN'): + self.fail(f'{name}: no golden image existed, so one was written to {path}. Review it and commit it.') + return + diff = np.abs(np.asarray(thumb, dtype=float) - np.asarray(Image.open(path).convert('RGB'), dtype=float)).mean() + self.assertLess(diff, self.TOLERANCE, f'{name} no longer matches tests/golden/{name}.png (diff {diff:.2f}). ' + f'If the change is intended, run: UPDATE_GOLDEN=1 python -m unittest tests.test_charts') + + def test_hbar_mixed(self): self.check('hbar_mixed') + def test_price_line(self): self.check('price_line') + def test_compare_lines(self): self.check('compare_lines') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_edges.py b/tests/test_edges.py new file mode 100644 index 00000000..8e3bbe50 --- /dev/null +++ b/tests/test_edges.py @@ -0,0 +1,125 @@ +"""Failure and no-data scenarios: empty or broken upstream data, bad requests, rejected callers.""" +import contextlib, io, json, unittest +from tests.common import * + +COMMANDS = ('.price', '.price 7d', '.price top', '.session', '.premarket', '.performance 7d', '.shorts', '.peg', '.buy', + '.sell', '.beta', '.pe', '.marketcap', '.earnings', '.dividend', '.trades', '.holdings', '.watchlist', + '.T1', '.history T1', '.compare T1 T2') + + +class UpstreamEmpty(FinbotCase): + """Yahoo answers, but with nothing usable. Every command must reply sanely and never crash.""" + + ONE_LINE_ERROR = ('.history T1', '.compare T1 T2') # these say so with a single error line when the ticker has no data + + def run_all(self): + for cmd in COMMANDS: + with self.subTest(cmd=cmd): + if cmd in self.ONE_LINE_ERROR: + self.sent.clear(); self.errors.clear() + worker.process_request('telegram', '55', '@u', cmd, BOT, 'U', '1') + self.assertLessEqual(len(self.errors), 1, cmd) + else: + self.command(cmd) + + def test_yahoo_returns_empty_dict(self): + self.patch(yahoo, 'fetch', lambda ts: {}) + self.patch(yahoo, 'fetch_detail', lambda t, *a, **k: {}) + self.run_all() + + def test_yahoo_raises(self): + def down(*a, **k): + raise RuntimeError('Yahoo API unavailable') + for name in ('fetch', 'fetch_detail', 'price_series'): + self.patch(yahoo, name, down) + for cmd in ('.price', '.peg', '.T1', '.compare T1 T2'): + with self.subTest(cmd=cmd): + self.sent.clear(); self.errors.clear() + worker.process_request('telegram', '55', '@u', cmd, BOT, 'U', '1') + self.assertEqual(len(self.errors), 1, cmd) # exactly one one-line report, not a traceback to the user + self.assertTrue(self.errors[0].startswith('⚠️')) + + def test_sharesight_down_is_reported_once(self): + def down(*a, **k): + raise RuntimeError('Sharesight performance error: boom') + self.patch(sharesight, 'get_performance_wrapper', down) + self.patch(sharesight, 'get_performance', down) + self.patch(sharesight, 'get_portfolios', down) + self.sent.clear(); self.errors.clear() + worker.process_request('telegram', '55', '@u', '.performance 7d', BOT, 'U', '1') + self.assertEqual(len(self.errors), 1) + + def test_no_holdings_and_empty_watchlist(self): + self.patch(util, 'get_holdings_and_watchlist', lambda: []) + for cmd in ('.watchlist', '.price', '.earnings'): + with self.subTest(cmd=cmd): + self.command(cmd) + + +class WatchlistActions(FinbotCase): + def test_add_duplicate_delete_and_unknown_action(self): + add = self.command('.watchlist add MSFT') + self.assertTrue(add) + self.assertEqual(util.json_load('finbot_watchlist.json', persist=True), ['MSFT']) + again = self.command('.watchlist add MSFT') + self.assertNotEqual(add, again) # a duplicate is reported differently from a first add + self.assertEqual(util.json_load('finbot_watchlist.json', persist=True), ['MSFT']) + self.command('.watchlist rm MSFT') + self.assertEqual(util.json_load('finbot_watchlist.json', persist=True), []) + self.assertIn('not a valid watchlist action', str(self.command('.watchlist frobnicate MSFT'))) + + +class BadArguments(FinbotCase): + def test_compare_argument_errors_give_usage(self): + for cmd in ('.compare T1', '.compare T1 T2 T3 T4 T5 T6 T7', '.compare T1 T1'): + with self.subTest(cmd=cmd): + self.assertIn('Usage', str(self.command(cmd))) + + def test_compare_with_a_ticker_that_has_no_history(self): + def no_history(t): + raise RuntimeError(f'no price history for {t}') + self.patch(yahoo, 'price_series', no_history) + self.sent.clear(); self.errors.clear() + worker.process_request('telegram', '55', '@u', '.compare T1 T2', BOT, 'U', '1') + self.assertEqual(len(self.errors), 1) + + def test_history_with_extra_argument_is_rejected_cleanly(self): + self.assertTrue(self.command('.history')) + self.assertTrue(self.command('.history T1 1y')) + + +class WsgiRejections(FinbotCase): + def request(self, path='/telegram', body=None, headers=None): + body = json.dumps(body if body is not None else {}).encode() + env = {'PATH_INFO': path, 'wsgi.input': io.BytesIO(body), **(headers or {})} + status = [] + with contextlib.redirect_stderr(io.StringIO()), contextlib.redirect_stdout(io.StringIO()): + out = bot.main(env, lambda s, *a, **k: status.append(s)) + return b''.join(out), status[-1] + + def test_telegram_requires_the_secret_header(self): + message = {'message': {'message_id': 1, 'chat': {'id': 5}, 'from': {'first_name': 'A'}, 'text': '.help'}} + self.assertIn(b'Unauthorized', self.request(body=message)[0]) + self.assertIn(b'Unauthorized', self.request(body=message, headers={'HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN': 'wrong'})[0]) + self.assertEqual(self.sent, []) # nothing was processed + + def test_unknown_path_is_404(self): + self.assertEqual(self.request(path='/nope')[1], '404 Not Found') + + def test_malformed_body_is_a_500_not_a_crash(self): + env = {'PATH_INFO': '/telegram', 'wsgi.input': io.BytesIO(b'not json')} + status = [] + with contextlib.redirect_stderr(io.StringIO()): + bot.main(env, lambda s, *a, **k: status.append(s)) + self.assertTrue(status[-1].startswith('500')) + + def test_telegram_edge_updates(self): + secret = {'HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN': 'tok'} + self.assertEqual(self.request(body={'edited_message': {}}, headers=secret)[0], b'Unsupported') + self.assertEqual(self.request(body={'callback_query': {'id': '1'}}, headers=secret)[0], b'Unsupported') + odd = {'message': {'message_id': 1, 'chat': {'id': 5}, 'from': {'first_name': 'A'}, 'new_chat_members': []}} + self.assertIn(b'Unhandled', self.request(body=odd, headers=secret)[0]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_finbot.py b/tests/test_finbot.py index f9c437b3..ce456425 100644 --- a/tests/test_finbot.py +++ b/tests/test_finbot.py @@ -1,87 +1,49 @@ -"""Offline smoke tests: external services (Telegram, Yahoo, Sharesight, Short Man) are mocked. - -Run with: python -m unittest discover -s tests -v -""" -import unittest -import sys, os, time, io, json, datetime, tempfile, contextlib -ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -sys.path.insert(0, ROOT) -os.environ.update(telegramOutgoingWebhook='https://example.com/telegram', telegramOutgoingToken='tok') -os.chdir(ROOT) +"""Plumbing: message chunking, timezones, the WSGI handler.""" +import contextlib, datetime, io, json, unittest, zoneinfo from unittest import mock -from lib import webhook, util, yahoo, sharesight, shortman, worker, charts, telegram, reports -import performance, shorts, price, reminder, cal -util.config_var_dir = tempfile.mkdtemp(); util.config_cache_dir = tempfile.mkdtemp() -webhook.webhooks.clear(); webhook.webhooks['telegram'] = 'https://api.telegram.org/botX/' -for m in (performance, shorts, price, cal): m.webhooks = webhook.webhooks +from tests.common import * -class Smoke(unittest.TestCase): - def check(self, name, cond): - self.assertTrue(cond, name) +class Plumbing(FinbotCase): + def test_chunked_send_keeps_keyboard_on_last_chunk(self): + posted = [] - def setUp(self): - fake = mock.Mock(status_code=200, text='{}') - fake.json.return_value = {'ok': True, 'result': {'message_id': 1, 'chat': {'id': 5}}} - patcher = mock.patch('requests.Session.request', return_value=fake) # no real network traffic - patcher.start() - self.addCleanup(patcher.stop) + class Response: + status_code = 200 + def json(self): return {'ok': True, 'result': {'message_id': len(posted), 'chat': {'id': 5}}} - def test_all(self): + def post(url, **kwargs): + posted.append(kwargs['json']) + return Response() - # 1. chunked send (the len(generator) bug) with a keyboard on the last chunk - posted = [] - class R: - status_code = 200 - def json(s): return {'ok': True, 'result': {'message_id': len(posted), 'chat': {'id': 5}}} - with mock.patch.object(webhook.requests, 'post', lambda url, **k: (posted.append(k['json']), R())[1]), mock.patch.object(webhook.time, 'sleep', lambda x: None): - res = webhook.payload_wrapper('telegram', 'https://api.telegram.org/botX/sendMessage?chat_id=5', ['x' * 60] * 300, '5', reply_markup={'keyboard': []}) - self.check(f'chunked send returns one result per chunk ({len(res)})', len(res) > 1 and len(posted) == len(res)) - self.check('keyboard only on the last chunk', 'reply_markup' in posted[-1] and all('reply_markup' not in p for p in posted[:-1])) + with mock.patch.object(webhook.requests, 'post', post), mock.patch.object(webhook.time, 'sleep', lambda s: None): + results = REAL_PAYLOAD_WRAPPER('telegram', 'https://api.telegram.org/botX/sendMessage?chat_id=5', + ['x' * 60] * 300, '5', reply_markup={'keyboard': []}) + self.assertGreater(len(results), 1) + self.assertEqual(len(posted), len(results)) + self.assertIn('reply_markup', posted[-1]) + self.assertTrue(all('reply_markup' not in p for p in posted[:-1])) - # 2. zoneinfo: reminder and yahoo timezone handling - tz = __import__('zoneinfo').ZoneInfo('Australia/Melbourne') - self.check('birthday tz offset is not LMT', datetime.datetime(2022, 8, 18, tzinfo=tz).utcoffset() == datetime.timedelta(hours=10)) - payload = None - with mock.patch.object(reminder, 'load_reminders', lambda: []): - reminder.config_timezone = 'Australia/Melbourne' - payload = reminder.lambda_handler.__code__ is not None - self.check('reminder imports/compiles with zoneinfo', payload) + def test_zoneinfo_offset_is_not_lmt(self): + tz = zoneinfo.ZoneInfo('Australia/Melbourne') + self.assertEqual(datetime.datetime(2022, 8, 18, tzinfo=tz).utcoffset(), datetime.timedelta(hours=10)) - # 3. reports through the worker on mocked data - sent = [] - webhook.sendPhoto = lambda chat, img, cap, svc, **k: sent.append(('photo', len(img.read()), cap.split('\n')[0])) - webhook.payload_wrapper = lambda svc, url, payload, *a, **k: sent.append(('text', len(payload))) - tick = [f'T{i}' for i in range(25)]; util.get_holdings_and_watchlist = lambda: tick - yahoo.fetch = lambda ts: {t: {'marketState': 'CLOSED', 'regularMarketTime': time.time(), 'profile_title': 'Co ' + t, 'profile_exchange': 'NasdaqGS', 'percent_change': 1.0, 'market_cap': 1e9, 'price_to_earnings_trailing': 9, 'price_to_earnings_forward': 9, 'beta': 2.0, 'short_percent': 20} for t in ts} - sharesight.get_performance_wrapper = lambda d: {1: {'report': {'holdings': [{'instrument': {'code': t, 'market_code': 'NASDAQ'}, 'capital_gain_percent': str(i * 4 - 40)} for i, t in enumerate(tick)]}}} - sharesight.get_portfolios = lambda: {'Rob': 1} - sharesight.get_performance = lambda pid, d: {'report': {'holdings': [{'portfolio': {'name': 'Rob'}}], 'currency_gain_percent': '0', 'capital_gain_percent': '2.5', 'total_gain_percent': '0'}} - yahoo.price_history = lambda t, days=None, **k: ({days: 1.5} if days else {'Max': 5, '1Y': 3}, None) - yahoo.prefetch_history = lambda ts: None - shortman.fetch = lambda md: md - yahoo.fetch_detail = lambda t, s=0: {t: dict(yahoo.fetch([t])[t], price_to_earnings_peg=1.2, recommend='buy', recommend_index=1.2, recommend_analysts=4)} - for label, call in ( - ('price top/bottom', lambda: price.lambda_handler(threshold=0, days=30, top=10)), - ('price list', lambda: price.lambda_handler(chat_id='5', threshold=3, service='telegram', interactive=True, days=30)), - ('performance', lambda: performance.lambda_handler(past_days=7)), - ('shorts', lambda: shorts.lambda_handler()), - ): - sent.clear(); call(); self.check(f'{label} -> {sent[0][0]} {sent[0][1:]}', sent and sent[0][0] == 'photo') - for cmd in ('.peg', '.buy', '.beta', '.help'): - sent.clear(); worker.process_request('telegram', '55', '@u', cmd, '@bot', 'U', '1'); self.check(f'{cmd} -> {sent[:1]}', bool(sent)) + def test_reminder_runs(self): + with mock.patch.object(reminder, 'load_reminders', lambda: []): + reminder.lambda_handler() - # 4. bot handler with real gevent: sticker is silent, unknown logged - import bot - def post_msg(message): - body = json.dumps({'update_id': 1, 'message': dict({'message_id': 5, 'chat': {'id': 55, 'type': 'private'}, 'from': {'id': 9, 'first_name': 'Rob'}, 'date': 1}, **message)}).encode() + def test_bot_handler(self): + def post(message): + body = json.dumps({'update_id': 1, 'message': dict({'message_id': 5, 'chat': {'id': 55, 'type': 'private'}, + 'from': {'id': 9, 'first_name': 'Rob'}, 'date': 1}, **message)}).encode() env = {'PATH_INFO': '/telegram', 'wsgi.input': io.BytesIO(body), 'HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN': 'tok'} err = io.StringIO() with contextlib.redirect_stderr(err): out = bot.main(env, lambda *a, **k: None) return out, err.getvalue() - self.check('sticker is silent', post_msg({'sticker': {'file_id': 'x'}}) == ([b''], '')) - self.check('git hash helper', len(bot.git_version()) >= 7) + self.assertEqual(post({'sticker': {'file_id': 'x'}}), ([b''], '')) # stickers: silent, not logged + self.assertGreaterEqual(len(bot.git_version()), 7) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_providers.py b/tests/test_providers.py new file mode 100644 index 00000000..4c02a3c8 --- /dev/null +++ b/tests/test_providers.py @@ -0,0 +1,263 @@ +"""Yahoo, Sharesight and Telegram parsing and failure handling, driven by the real API responses saved under doc/. +HTTP is mocked at the requests layer, so the production parsing code runs unchanged.""" +import copy, json, os, unittest +from unittest import mock +from tests.common import * + +DOC = os.path.join(ROOT, 'doc') + + +def sample(*path): + with open(os.path.join(DOC, *path), encoding='utf-8') as f: + return json.load(f) + + +class Resp: + def __init__(self, payload=None, status=200, text=''): + self.payload, self.status_code, self.text = payload, status, text or json.dumps(payload) + def json(self): + if self.payload is None: + raise ValueError('no json') + return self.payload + + +def serve(*responses): + """requests.get replacement: replays the responses in order (exceptions are raised); the last one repeats.""" + queue = list(responses) + calls = [] + def get(url, **kwargs): + calls.append(url) + item = queue.pop(0) if len(queue) > 1 else queue[0] + if isinstance(item, Exception): + raise item + return item + get.calls = calls + return get + + +class ProviderCase(FinbotCase): + def setUp(self): + super().setUp() + # these tests exercise the real parsers, so undo the canned data from the base fixture + for module, name in ((yahoo, 'fetch'), (yahoo, 'fetch_detail'), (yahoo, 'price_series'), (sharesight, 'get_portfolios'), + (sharesight, 'get_trades'), (sharesight, 'get_performance'), (sharesight, 'get_performance_wrapper')): + self.patch(module, name, REAL[(module.__name__, name)]) + self.patch(yahoo, 'getCookie', lambda *a, **k: 'A=B') + self.patch(yahoo, 'getCrumb', lambda *a, **k: 'crumb') + self.patch(sharesight, 'get_token', lambda: 'token') + self.patch(sharesight.time, 'sleep', lambda s: None) + + def http(self, module, *responses): + getter = serve(*responses) + self.patch(module.requests, 'get', getter) + return getter + + +class YahooQuotes(ProviderCase): + def test_parses_real_v7_quote(self): + self.http(yahoo, Resp(sample('yahoo', 'sample_yahoo_v7.json'))) + data = yahoo.fetch(['AAPL'])['AAPL'] + self.assertEqual(data['profile_title'], 'Apple') + self.assertEqual(data['percent_change'], 0.09) + self.assertEqual(data['percent_change_postmarket'], -0.13) + self.assertEqual(data['quoteType'], 'EQUITY') + self.assertEqual(data['marketState'], 'CLOSED') + self.assertEqual(data['profile_exchange'], 'NasdaqGS') + self.assertEqual(data['exchangeTimezoneName'], 'America/New_York') + self.assertGreater(data['market_cap'], 1e12) + + def test_second_call_is_served_from_cache(self): + getter = self.http(yahoo, Resp(sample('yahoo', 'sample_yahoo_v7.json'))) + yahoo.fetch(['AAPL']) + yahoo.fetch(['AAPL']) + self.assertEqual(len(getter.calls), 1) + + def test_falls_back_to_second_host(self): + getter = self.http(yahoo, Resp(status=500), Resp(sample('yahoo', 'sample_yahoo_v7.json'))) + self.assertIn('AAPL', yahoo.fetch(['AAPL'])) + self.assertIn('query1', getter.calls[1]) + + def test_all_hosts_failing_raises(self): + for failure in (Resp(status=500), ConnectionError('down')): + with self.subTest(failure=failure): + self.http(yahoo, failure) + with self.assertRaises(RuntimeError): + yahoo.fetch(['AAPL']) + + def test_empty_result_is_none(self): + self.http(yahoo, Resp({'quoteResponse': {'result': None, 'error': {'code': 'x'}}})) + self.assertEqual(yahoo.fetch(['NOPE']), {}) + + def test_no_tickers_is_none(self): + self.assertEqual(yahoo.fetch([]), {}) + + def test_incomplete_quotes_are_skipped(self): + body = sample('yahoo', 'sample_yahoo_v7.json') + good = body['quoteResponse']['result'][0] + no_price = {k: v for k, v in good.items() if k != 'regularMarketPrice'} | {'symbol': 'NOPRICE'} + no_name = {k: v for k, v in good.items() if k not in ('longName', 'shortName')} | {'symbol': 'NONAME'} + body['quoteResponse']['result'] = [good, no_price, no_name] + self.http(yahoo, Resp(body)) + self.assertEqual(list(yahoo.fetch(['AAPL', 'NOPRICE', 'NONAME'])), ['AAPL']) + + +class YahooDetail(ProviderCase): + @staticmethod + def combined(): + """quoteSummary as Yahoo returns it for several modules at once, assembled from the per-module samples.""" + merged = {} + for module in ('price', 'summaryProfile', 'summaryDetail', 'defaultKeyStatistics', 'financialData', 'calendarEvents', 'earnings'): + merged.update(sample('yahoo', f'sample_yahoo_v10_{module}.json')['quoteSummary']['result'][0]) + return {'quoteSummary': {'result': [merged], 'error': None}} + + def test_parses_real_quote_summary(self): + self.http(yahoo, Resp(self.combined())) + data = yahoo.fetch_detail('NVDA')['NVDA'] + self.assertEqual(data['profile_title'], 'NVIDIA') + self.assertTrue(data['profile_bio']) + + def test_http_failure_gives_empty(self): + for failure in (Resp(status=404), ConnectionError('down')): + with self.subTest(failure=failure): + self.http(yahoo, failure) + self.assertEqual(yahoo.fetch_detail('DRNA'), {}) + + def test_missing_names_give_empty(self): + body = self.combined() + body['quoteSummary']['result'][0]['price'].update(longName=None, shortName=None) + self.http(yahoo, Resp(body)) + self.assertEqual(yahoo.fetch_detail('DUB'), {}) + self.http(yahoo, Resp({'quoteSummary': {'result': None, 'error': {}}})) + self.assertEqual(yahoo.fetch_detail('GONE'), {}) + + +class YahooChart(ProviderCase): + def test_real_chart_to_dataframe_and_basics(self): + chart = sample('yahoo', 'sample_yahoo_v8_chart.json') + df = yahoo.chart_json_to_df(chart) + self.assertEqual(len(df), 2) + self.assertTrue({'Date', 'Close'} <= set(df.columns)) + basics = yahoo.chart_json_to_stock_basics(chart) + self.assertEqual((basics['symbol'], basics['currency']), ('CRWD', 'USD')) + + def test_fetch_chart_json_failures_return_error_strings(self): + for failure in (Resp(status=404), ConnectionError('down')): + with self.subTest(failure=failure): + self.http(yahoo, failure) + result = yahoo.fetch_chart_json('AAPL', full=True) + self.assertIsInstance(result, tuple) + self.assertIsNone(result[1]) + + def test_price_series_raises_on_failure_and_on_empty_history(self): + self.http(yahoo, Resp(status=500)) + with self.assertRaises(RuntimeError): + yahoo.price_series('AAPL') + empty = sample('yahoo', 'sample_yahoo_v8_chart.json') + empty['chart']['result'][0]['indicators']['quote'][0]['close'] = [None, None] + self.http(yahoo, Resp(empty)) + with self.assertRaises(RuntimeError): + yahoo.price_series('EMPTY') + + +class SharesightApi(ProviderCase): + def test_portfolios_from_real_response(self): + self.http(sharesight, Resp(sample('sharesight', 'finbot_sharesight_portfolios.json'))) + self.assertEqual(sharesight.get_portfolios(), {'Rob': 445825, 'RobSMSF': 684141}) + + def test_portfolio_exclusion_and_inclusion(self): + self.http(sharesight, Resp(sample('sharesight', 'finbot_sharesight_portfolios.json'))) + self.patch(sharesight, 'config_exclude_portfolios', ['445825']) + self.assertEqual(sharesight.get_portfolios(), {'RobSMSF': 684141}) + self.patch(sharesight, 'config_exclude_portfolios', []) + self.patch(sharesight, 'config_include_portfolios', ['445825']) + self.assertEqual(sharesight.get_portfolios(), {'Rob': 445825}) + + def test_portfolio_failures_raise(self): + for failure in (Resp({'error': 'nope', 'error_code': 'x'}, 401), ConnectionError('down'), Resp({'portfolios': []})): + with self.subTest(failure=failure): + self.http(sharesight, failure) + with self.assertRaises(RuntimeError): + sharesight.get_portfolios() + + def test_trades_from_real_response(self): + self.http(sharesight, Resp(sample('sharesight', 'finbot_sharesight_trades_684141_1.json'))) + trades = sharesight.get_trades('RobSMSF', 684141, 1) + self.assertEqual(len(trades), 2) + self.assertEqual(trades[0]['portfolio'], 'RobSMSF') + self.assertEqual(trades[0]['symbol'], 'FANG') + + def test_trade_failures_raise(self): + self.http(sharesight, ConnectionError('down')) + with self.assertRaises(RuntimeError): + sharesight.get_trades('RobSMSF', 684141, 1) + self.http(sharesight, Resp({'trades': [], 'error': 'bad', 'error_code': 'x'}, 400)) + with self.assertRaises(RuntimeError): + sharesight.get_trades('RobSMSF', 684141, 2) + + def test_performance_and_holdings_from_real_response(self): + self.http(sharesight, Resp(sample('sharesight', 'finbot_sharesight_performance_684141_0.json'))) + self.assertEqual(len(sharesight.get_performance(684141, 0)['report']['holdings']), 18) + tickers = sharesight.get_holdings('RobSMSF', 684141) + self.assertIn('ARM', tickers) # US venues carry no Yahoo suffix + self.assertEqual(tickers, sorted(tickers)) + + def test_performance_retries_when_sharesight_is_busy(self): + busy = Resp({'error': 'Too many parallel requests', 'error_code': 'x'}, 429) + getter = self.http(sharesight, busy, busy, Resp(sample('sharesight', 'finbot_sharesight_performance_684141_0.json'))) + self.assertIn('report', sharesight.get_performance(684141, 0)) + self.assertEqual(len(getter.calls), 3) + + def test_performance_gives_up_after_repeated_busy_or_hard_errors(self): + for failure in (Resp({'error': 'Too many parallel requests', 'error_code': 'x'}, 429), + Resp({'error': 'forbidden', 'error_code': 'x'}, 403), ConnectionError('down')): + with self.subTest(failure=failure): + self.http(sharesight, failure) + with self.assertRaises(RuntimeError): + sharesight.get_performance(684141, 7) + + +class TelegramSending(FinbotCase): + """webhook.write against failing endpoints. The real send path, not the base fixture's recorder.""" + + def post(self, outcome): + def fake(url, **kwargs): + if isinstance(outcome, Exception): + raise outcome + return outcome + self.patch(webhook.requests, 'post', fake) + return webhook.write('telegram', 'https://api.telegram.org/botX/sendMessage?chat_id=5', 'hi', '5') + + def test_success_returns_json(self): + self.assertEqual(self.post(Resp({'ok': True})), {'ok': True}) + + def test_http_error_and_exception_return_none(self): + self.assertIsNone(self.post(Resp(status=429))) + self.assertIsNone(self.post(ConnectionError('down'))) + + def test_non_json_success_returns_empty_dict(self): + self.assertEqual(self.post(Resp(None)), {}) + + def test_error_line_redacts_secrets_and_truncates(self): + line = webhook.error_line(RuntimeError('GET /bot123456:ABC-def_1/send?crumb=SECRET&x=1 failed'), 'ctx') + self.assertNotIn('ABC-def_1', line) + self.assertNotIn('SECRET', line) + self.assertTrue(line.startswith('⚠️ ctx: ')) + self.assertLessEqual(len(webhook.error_line('x' * 1000)), 205) + + def test_report_error_never_raises(self): + self.patch(webhook, 'write', mock.Mock(side_effect=RuntimeError('send failed'))) + webhook.report_error(ValueError('boom'), service='telegram', chat_id='5') + webhook.report_error(ValueError('boom')) + + def test_guarded_reports_and_exits_nonzero(self): + reported = [] + self.patch(webhook, 'report_error', lambda e, **k: reported.append(e)) + with self.assertRaises(SystemExit) as raised: + webhook.guarded('job', lambda: 1 / 0) + self.assertEqual(raised.exception.code, 1) + self.assertEqual(len(reported), 1) + self.assertEqual(webhook.guarded('job', lambda: 'fine'), 'fine') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_reports.py b/tests/test_reports.py new file mode 100644 index 00000000..10b41334 --- /dev/null +++ b/tests/test_reports.py @@ -0,0 +1,112 @@ +"""Every chat command and scheduled report produces output on mocked data.""" +import unittest +from tests.common import * + + +class Commands(FinbotCase): + def test_text_commands(self): + for cmd in ('.help', '/start', '.hello', '.thanks', '.who', '.plan', '.holdings', '.marketcap', + '.pe', '.fpe', '.peg', '.beta', '.buy', '.sell', '.earnings', '.dividend', + '.earnings T1', '.dividend T1', '.trades', '.trades 30d', '.price T1', '.session T1', + '.premarket T1', '.shorts T1', '.watchlist', '.T1'): + with self.subTest(cmd=cmd): + self.assertTrue(self.command(cmd), f'{cmd} sent nothing') + + def test_chart_commands(self): + for cmd in ('.price', '.price 7d', '.price top', '.price bottom', '.session', '.premarket', '.postmarket', + '.performance', '.performance 7d', '.performance 1y', '.shorts', '.peg', '.buy', '.sell', '.beta', '.top10 week'): + with self.subTest(cmd=cmd): + out = self.command(cmd) + self.assertTrue(out, f'{cmd} sent nothing') + self.assertTrue(any(kind == 'photo' for kind, *_ in out) or cmd.startswith('.top10'), f'{cmd} sent no chart: {out}') + + def test_history_and_compare(self): + for cmd in ('.history T1', '.compare T1 T2', '.compare T1 T2 3m'): + with self.subTest(cmd=cmd): + self.assertTrue(self.command(cmd), f'{cmd} sent nothing') + + def test_trades_and_rating_report_when_nothing_is_stored(self): + # first run: no state files exist yet, which used to crash these + for cmd in ('.trades', '.watchlist'): + self.assertTrue(self.command(cmd), cmd) + + def test_unknown_stickers_and_noise_are_ignored(self): + self.assertEqual(self.command('lorem ipsum dolor sit amet'), []) + + +class ReportFunctions(FinbotCase): + def test_prepare_help(self): + self.assertTrue(reports.prepare_help('telegram', BOT)) + + def test_prepare_holdings(self): + self.assertTrue(reports.prepare_holdings_payload(None, 'telegram', '@u')) + + def test_prepare_marketcap(self): + self.assertTrue(reports.prepare_marketcap_payload('telegram', 'top', length=5)) + + def test_prepare_rating(self): + for action in ('buy', 'sell'): + self.assertTrue(reports.prepare_rating_payload('telegram', action, length=5)) + + def test_prepare_value(self): + for action in ('pe', 'fpe', 'peg', 'beta'): + with self.subTest(action=action): + self.assertTrue(reports.prepare_value_payload('telegram', action, length=5)) + + def test_prepare_profile(self): + self.assertTrue(reports.prepare_profile_payload('telegram', '@u', 'T1')) + + def test_prepare_watchlist(self): + self.assertTrue(reports.prepare_watchlist('telegram', '@u')) + + def test_prepare_compare_usage(self): + caption = reports.prepare_compare('telegram', ['T1'])[0] + self.assertIn('Usage', ' '.join(caption)) + + +class ScheduledReports(FinbotCase): + """The cron entry points (non-interactive: they post to the default channel).""" + + def test_price(self): + price.lambda_handler() + self.assertTrue(self.sent) + + def test_performance(self): + performance.lambda_handler(past_days=7) + self.assertTrue(self.sent) + + def test_shorts(self): + shorts.lambda_handler() + self.assertTrue(self.sent) + + def test_earnings_and_dividends(self): + for kwargs in ({'earnings': True}, {'dividend': True}): + with self.subTest(kwargs=kwargs): + self.sent.clear() + cal.lambda_handler(**kwargs) + self.assertTrue(self.sent) + + def test_trades(self): + trades.lambda_handler(interactive=True, service='telegram', chat_id='5') + self.assertTrue(self.sent) + + def test_milestone(self): + milestone.lambda_handler(interactive=True, service='telegram', chat_id='5') + self.assertTrue(self.sent) + + def test_rating(self): + rating.lambda_handler(interactive=True, service='telegram', chat_id='5') + self.assertTrue(self.sent) + + +class Podcasts(unittest.TestCase): + def test_format_helpers(self): + import podcasts + self.assertEqual(podcasts.format_duration('3725'), '01:02') + self.assertEqual(podcasts.format_duration('1:02:03'), '01:02') + self.assertEqual(podcasts.format_duration('nonsense'), '') + self.assertTrue(podcasts.how_long_ago('Mon, 01 Jan 2024 00:00:00 +0000').endswith('d ago')) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_util.py b/tests/test_util.py new file mode 100644 index 00000000..22a3afd2 --- /dev/null +++ b/tests/test_util.py @@ -0,0 +1,98 @@ +"""Pure helpers in lib/util.py: table-driven pass and fail cases.""" +import datetime, unittest +from tests.common import * + + +class Parsing(unittest.TestCase): + def test_days_from_human_days(self): + for arg, days in (('7', 7), ('7d', 7), ('7D', 7), ('2w', 14), ('3m', 90), ('1y', 365), ('0', 0)): + with self.subTest(arg=arg): + self.assertEqual(util.days_from_human_days(arg), days) + today = datetime.date.today() + self.assertEqual(util.days_from_human_days('ytd'), (today - today.replace(month=1, day=1)).days) + + def test_days_from_human_days_rejects_non_periods(self): + for arg in ('AAPL', '', 'x5d', '1.5y', 'week'): + with self.subTest(arg=arg), self.assertRaises(ValueError): + util.days_from_human_days(arg) + + def test_transform_to_yahoo(self): + for args, expected in ((('AAPL',), 'AAPL'), (('FANG', 'ASX'), 'FANG.AX'), (('ARM', 'NASDAQ'), 'ARM'), + (('ARM:NASDAQ',), 'ARM'), (('BHP.AX',), 'BHP.AX'), (('0700', 'HKG'), '0700.HK'), + (('BRK', 'B'), 'BRK-B'), (('2330', 'TAI'), '2330.TW'), (('VOD', 'LSE'), 'VOD.L')): + with self.subTest(args=args): + self.assertEqual(util.transform_to_yahoo(*args), expected) + + def test_transform_title(self): + self.assertEqual(util.transform_title('APPLE INC'), 'Apple') + self.assertEqual(util.transform_title('Acme FPO'), 'Acme') + self.assertEqual(util.transform_title('Mixed Case Name'), 'Mixed Case Name') + + +class Formatting(unittest.TestCase): + def test_signed_percent(self): + for value, decimals, expected in ((12, None, '+12%'), (-3.5, None, '-3.5%'), (0, None, '0%'), (1234, None, '+1,234%'), + (2.345, 1, '+2.3%'), (-0.04, 1, '-0.0%')): + with self.subTest(value=value): + self.assertEqual(util.signed_percent(value, decimals), expected) + + def test_days_english(self): + for days, expected in ((None, 'today'), (0, 'today'), (1, 'the past day'), (7, 'the past week')): + with self.subTest(days=days): + self.assertEqual(util.days_english(days), expected) + + def test_date_range_english(self): + end = datetime.date(2026, 10, 5) + self.assertEqual(util.date_range_english(7, end), '28 Sep - 5 Oct 2026') + self.assertEqual(util.date_range_english(30, end), '5 Sep - 5 Oct 2026') + self.assertEqual(util.date_range_english(10, datetime.date(2026, 1, 5)), '26 Dec 2025 - 5 Jan 2026') # crosses a year + with self.assertRaises(TypeError): + util.date_range_english(None, end) # callers must pass a period + + def test_ordinal(self): + for number, suffix in ((1, 'st'), (2, 'nd'), (3, 'rd'), (4, 'th'), (11, 'th'), (12, 'th'), (13, 'th'), (21, 'st'), (112, 'th')): + with self.subTest(number=number): + self.assertEqual(util.ordinal(number), suffix) + + def test_get_emoji(self): + self.assertEqual([util.get_emoji(n) for n in (1, -1, 0)], ['🔼', '🔻', '▪️']) + + def test_links_per_service(self): + telegram_link = util.link('https://example.com/x', 'text', 'telegram') + slack_link = util.link('https://example.com/x', 'text', 'slack') + self.assertIn('