# -*- coding: utf-8 -*- from __future__ import unicode_literals import os import curses import logging import threading from functools import partial import praw import pytest from vcr import VCR from six.moves.urllib.parse import urlparse, parse_qs from six.moves.BaseHTTPServer import HTTPServer from rtv.oauth import OAuthHelper, OAuthHandler from rtv.config import Config from rtv.terminal import Terminal from rtv.subreddit import SubredditPage from rtv.submission import SubmissionPage from rtv.subscription import SubscriptionPage try: from unittest import mock except ImportError: import mock # Turn on autospec by default for convenience patch = partial(mock.patch, autospec=True) # Turn on logging, but disable vcr from spamming logging.basicConfig( level=logging.DEBUG, format='%(asctime)s:%(levelname)s:%(filename)s:%(lineno)d:%(message)s') for name in ['vcr.matchers', 'vcr.stubs']: logging.getLogger(name).disabled = True def pytest_addoption(parser): parser.addoption('--record-mode', dest='record_mode', default='none') parser.addoption('--refresh-token', dest='refresh_token', default='tests/refresh-token') class MockStdscr(mock.MagicMock): """ Extend mock to mimic curses.stdscr by keeping track of the terminal coordinates and allowing for the creation of subwindows with the same properties as stdscr. """ def getyx(self): return self.y, self.x def getmaxyx(self): return self.nlines, self.ncols def derwin(self, *args): """ derwin() derwin(begin_y, begin_x) derwin(nlines, ncols, begin_y, begin_x) """ if 'subwin' not in dir(self): self.attach_mock(MockStdscr(), 'subwin') if len(args) == 0: nlines = self.nlines ncols = self.ncols elif len(args) == 2: nlines = self.nlines - args[0] ncols = self.ncols - args[1] else: nlines = min(self.nlines - args[2], args[0]) ncols = min(self.ncols - args[3], args[1]) self.subwin.nlines = nlines self.subwin.ncols = ncols self.subwin.x = 0 self.subwin.y = 0 return self.subwin @pytest.fixture(scope='session') def vcr(request): def auth_matcher(r1, r2): return (r1.headers.get('authorization') == r2.headers.get('authorization')) def uri_with_query_matcher(r1, r2): "URI matcher that allows query params to appear in any order" p1, p2 = urlparse(r1.uri), urlparse(r2.uri) return (p1[:3] == p2[:3] and parse_qs(p1.query, True) == parse_qs(p2.query, True)) # Use `none` to use the recorded requests, and `once` to delete existing # cassettes and re-record. record_mode = request.config.option.record_mode assert record_mode in ('once', 'none') cassette_dir = os.path.join(os.path.dirname(__file__), 'cassettes') if not os.path.exists(cassette_dir): os.makedirs(cassette_dir) # https://github.com/kevin1024/vcrpy/pull/196 vcr = VCR( record_mode=request.config.option.record_mode, filter_headers=[('Authorization', '**********')], filter_post_data_parameters=[('refresh_token', '**********')], match_on=['method', 'uri_with_query', 'auth', 'body'], cassette_library_dir=cassette_dir) vcr.register_matcher('auth', auth_matcher) vcr.register_matcher('uri_with_query', uri_with_query_matcher) return vcr @pytest.fixture() def refresh_token(request): if request.config.option.record_mode == 'none': return 'mock_refresh_token' else: return open(request.config.option.refresh_token).read() @pytest.yield_fixture() def config(): conf = Config() with mock.patch.object(conf, 'save_history'), \ mock.patch.object(conf, 'delete_history'), \ mock.patch.object(conf, 'save_refresh_token'), \ mock.patch.object(conf, 'delete_refresh_token'): def delete_refresh_token(): # Skip the os.remove conf.refresh_token = None conf.delete_refresh_token.side_effect = delete_refresh_token yield conf @pytest.yield_fixture() def stdscr(): with patch('curses.initscr'), \ patch('curses.echo'), \ patch('curses.flash'), \ patch('curses.endwin'), \ patch('curses.newwin'), \ patch('curses.noecho'), \ patch('curses.cbreak'), \ patch('curses.doupdate'), \ patch('curses.nocbreak'), \ patch('curses.curs_set'), \ patch('curses.init_pair'), \ patch('curses.color_pair'), \ patch('curses.start_color'), \ patch('curses.use_default_colors'): out = MockStdscr(nlines=40, ncols=80, x=0, y=0) curses.initscr.return_value = out curses.newwin.side_effect = lambda *args: out.derwin(*args) curses.color_pair.return_value = 23 curses.ACS_VLINE = 0 yield out @pytest.yield_fixture() def reddit(vcr, request): cassette_name = '%s.yaml' % request.node.name # Clear the cassette before running the test if request.config.option.record_mode == 'once': filename = os.path.join(vcr.cassette_library_dir, cassette_name) if os.path.exists(filename): os.remove(filename) with vcr.use_cassette(cassette_name): with patch('praw.Reddit.get_access_information'): reddit = praw.Reddit(user_agent='rtv test suite', decode_html_entities=False, disable_update_check=True) if request.config.option.record_mode == 'none': # Turn off praw rate limiting when using cassettes reddit.config.api_request_delay = 0 yield reddit @pytest.fixture() def terminal(stdscr, config): term = Terminal(stdscr, config=config) # Disable the python 3.4 addch patch so that the mock stdscr calls are # always made the same way term.addch = lambda window, *args: window.addch(*args) return term @pytest.fixture() def oauth(reddit, terminal, config): return OAuthHelper(reddit, terminal, config) @pytest.yield_fixture() def oauth_server(): # Start the OAuth server on a random port in the background server = HTTPServer(('', 0), OAuthHandler) server.url = 'http://{0}:{1}/'.format(*server.server_address) thread = threading.Thread(target=server.serve_forever) thread.start() try: yield server finally: server.shutdown() thread.join() server.server_close() @pytest.fixture() def submission_page(reddit, terminal, config, oauth): submission = 'https://www.reddit.com/r/Python/comments/2xmo63' with terminal.loader(): page = SubmissionPage(reddit, terminal, config, oauth, url=submission) assert terminal.loader.exception is None page.draw() return page @pytest.fixture() def subreddit_page(reddit, terminal, config, oauth): subreddit = '/r/python' with terminal.loader(): page = SubredditPage(reddit, terminal, config, oauth, subreddit) assert not terminal.loader.exception page.draw() return page @pytest.fixture() def subscription_page(reddit, terminal, config, oauth): content_type = 'popular' with terminal.loader(): page = SubscriptionPage(reddit, terminal, config, oauth, content_type) assert terminal.loader.exception is None page.draw() return page