<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import json
import logging
import os.path
import unittest

from app import app
from datetime import datetime
from jobs import (avinor, calendar, flybussen, hackernews, nsb, ping, rss,
                  vaernesekspressen, yr)
from multiprocessing import Process, set_start_method
from requests import Session
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from tempfile import mkdtemp
from util.create_dashboard import DashboardFactory
from util.create_widget import WidgetFactory
from werkzeug.serving import run_simple
from http.server import BaseHTTPRequestHandler, HTTPServer


class TestRequestHandler(BaseHTTPRequestHandler):

    def _response(self):
        return self.server.test_responses.get(self.command, {}).get(self.path)

    def _write_response(self):
        response = self._response()
        status_code = 200
        if response is None:
            status_code = 404
            response = {'error': 'No match for {} {}'.format(self.command,
                                                             self.path)}
        self.send_response(status_code)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))

    def do_GET(self):
        self._write_response()

    def do_POST(self):
        self._write_response()

    def log_message(self, format, *args):
        # Disable request logging
        return


def test_data(name, parse_json=False):
    path = os.path.join(os.path.dirname(__file__), 'test_data', name)
    with open(path, 'rb') as f:
        if parse_json:
            return json.loads(f.read().decode('utf-8'))
        else:
            return f.read()


class App(unittest.TestCase):

    def setUp(self):
        app.testing = True
        app.debug = True
        app.config['DEFAULT_LAYOUT'] = None
        app.config['JOBS'] = {
            'mock': {
                'enabled': True,
                'interval': 60
            }
        }
        app.logger.setLevel(logging.WARN)
        logging.getLogger('werkzeug').setLevel(logging.WARN)
        # Run the application in a separate process so that streamed responses
        # are not blocked on running the client in the same thread
        self.p = Process(target=run_simple, args=('127.0.0.1', 8080, app),
                         kwargs={'use_debugger': app.debug})
        self.p.start()

    def url(self, path):
        return 'http://127.0.0.1:8080' + path

    def session(self):
        s = Session()
        s.mount('http://', HTTPAdapter(
            max_retries=Retry(total=100, backoff_factor=0.1))
        )
        return s

    def get(self, path, **kwargs):
        return self.session().get(self.url(path), **kwargs)

    def post(self, path, **kwargs):
        return self.session().post(self.url(path), **kwargs)

    def test_widget(self):
        r = self.get('/widget/foo')
        self.assertEqual(404, r.status_code)

        r = self.get('/widget/mock')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

        r = self.get('/w/mock')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

    def test_dashboard(self):
        r = self.get('/dashboard/foo')
        self.assertEqual(404, r.status_code)

        r = self.get('/')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

    def test_events(self):
        r = self.get('/events', stream=True)
        with r:
            data = next(r.iter_lines(chunk_size=1)).decode('utf-8')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/event-stream; charset=utf-8',
                         r.headers['content-type'])
        self.assertEqual('data: {"body":{"data":"spam"},"job":"mock"}', data)

    def test_events_post(self):
        r = self.post('/events/foo')
        self.assertEqual(404, r.status_code)

        r = self.post('/events/mock', json={'data': 'eggs'})
        self.assertEqual(201, r.status_code)

        r = self.get('/events', stream=True)
        with r:
            data = next(r.iter_lines(chunk_size=1)).decode('utf-8')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/event-stream; charset=utf-8',
                         r.headers['content-type'])
        self.assertEqual('data: {"body":{"data":"eggs"},"job":"mock"}', data)

    def tearDown(self):
        self.p.terminate()
        self.p.join()


class Yr(unittest.TestCase):

    def setUp(self):
        self.json = test_data('varsel.json', parse_json=True)
        self.yr = yr.Yr({'interval': None, 'url': None,
                         'location': 'Trondheim'})

    def test_parse_tree_new_api(self):
        data = self.yr._parse(self.json, datetime(2020, 8, 17, 15))
        self.assertEqual('Trondheim', data['today']['location'])
        self.assertEqual(19.9, data['today']['temperature'])
        self.assertEqual('Klarvær', data['today']['description'])
        self.assertEqual('nordvest', data['today']['wind']['direction'])
        self.assertEqual(2.8, data['today']['wind']['speed'])

        self.assertEqual('Trondheim', data['tomorrow']['location'])
        self.assertEqual(23.0, data['tomorrow']['temperature'])
        self.assertEqual('Klarvær', data['tomorrow']['description'])
        self.assertEqual('nordvest', data['tomorrow']['wind']['direction'])
        self.assertEqual(2.2, data['tomorrow']['wind']['speed'])


class HackerNews(unittest.TestCase):

    def setUp(self):
        self.html = test_data('hn.html')
        self.hn = hackernews.HackerNews({'interval': None})

    def test_parse(self):
        data = self.hn._parse(self.html)

        self.assertEqual(30, len(data['items']))
        self.assertTrue('points' in data['items'][3])
        self.assertTrue('title' in data['items'][3])
        self.assertEqual(76, data['items'][3]['points'])
        self.assertEqual('Building an OpenBSD Router',<commit_msg>yr: Change test method name<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import json
import logging
import os.path
import unittest

from app import app
from datetime import datetime
from jobs import (avinor, calendar, flybussen, hackernews, nsb, ping, rss,
                  vaernesekspressen, yr)
from multiprocessing import Process, set_start_method
from requests import Session
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from tempfile import mkdtemp
from util.create_dashboard import DashboardFactory
from util.create_widget import WidgetFactory
from werkzeug.serving import run_simple
from http.server import BaseHTTPRequestHandler, HTTPServer


class TestRequestHandler(BaseHTTPRequestHandler):

    def _response(self):
        return self.server.test_responses.get(self.command, {}).get(self.path)

    def _write_response(self):
        response = self._response()
        status_code = 200
        if response is None:
            status_code = 404
            response = {'error': 'No match for {} {}'.format(self.command,
                                                             self.path)}
        self.send_response(status_code)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))

    def do_GET(self):
        self._write_response()

    def do_POST(self):
        self._write_response()

    def log_message(self, format, *args):
        # Disable request logging
        return


def test_data(name, parse_json=False):
    path = os.path.join(os.path.dirname(__file__), 'test_data', name)
    with open(path, 'rb') as f:
        if parse_json:
            return json.loads(f.read().decode('utf-8'))
        else:
            return f.read()


class App(unittest.TestCase):

    def setUp(self):
        app.testing = True
        app.debug = True
        app.config['DEFAULT_LAYOUT'] = None
        app.config['JOBS'] = {
            'mock': {
                'enabled': True,
                'interval': 60
            }
        }
        app.logger.setLevel(logging.WARN)
        logging.getLogger('werkzeug').setLevel(logging.WARN)
        # Run the application in a separate process so that streamed responses
        # are not blocked on running the client in the same thread
        self.p = Process(target=run_simple, args=('127.0.0.1', 8080, app),
                         kwargs={'use_debugger': app.debug})
        self.p.start()

    def url(self, path):
        return 'http://127.0.0.1:8080' + path

    def session(self):
        s = Session()
        s.mount('http://', HTTPAdapter(
            max_retries=Retry(total=100, backoff_factor=0.1))
        )
        return s

    def get(self, path, **kwargs):
        return self.session().get(self.url(path), **kwargs)

    def post(self, path, **kwargs):
        return self.session().post(self.url(path), **kwargs)

    def test_widget(self):
        r = self.get('/widget/foo')
        self.assertEqual(404, r.status_code)

        r = self.get('/widget/mock')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

        r = self.get('/w/mock')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

    def test_dashboard(self):
        r = self.get('/dashboard/foo')
        self.assertEqual(404, r.status_code)

        r = self.get('/')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/html; charset=utf-8', r.headers['content-type'])

    def test_events(self):
        r = self.get('/events', stream=True)
        with r:
            data = next(r.iter_lines(chunk_size=1)).decode('utf-8')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/event-stream; charset=utf-8',
                         r.headers['content-type'])
        self.assertEqual('data: {"body":{"data":"spam"},"job":"mock"}', data)

    def test_events_post(self):
        r = self.post('/events/foo')
        self.assertEqual(404, r.status_code)

        r = self.post('/events/mock', json={'data': 'eggs'})
        self.assertEqual(201, r.status_code)

        r = self.get('/events', stream=True)
        with r:
            data = next(r.iter_lines(chunk_size=1)).decode('utf-8')
        self.assertEqual(200, r.status_code)
        self.assertEqual('text/event-stream; charset=utf-8',
                         r.headers['content-type'])
        self.assertEqual('data: {"body":{"data":"eggs"},"job":"mock"}', data)

    def tearDown(self):
        self.p.terminate()
        self.p.join()


class Yr(unittest.TestCase):

    def setUp(self):
        self.json = test_data('varsel.json', parse_json=True)
        self.yr = yr.Yr({'interval': None, 'url': None,
                         'location': 'Trondheim'})

    def test_parse(self):
        data = self.yr._parse(self.json, datetime(2020, 8, 17, 15))
        self.assertEqual('Trondheim', data['today']['location'])
        self.assertEqual(19.9, data['today']['temperature'])
        self.assertEqual('Klarvær', data['today']['description'])
        self.assertEqual('nordvest', data['today']['wind']['direction'])
        self.assertEqual(2.8, data['today']['wind']['speed'])

        self.assertEqual('Trondheim', data['tomorrow']['location'])
        self.assertEqual(23.0, data['tomorrow']['temperature'])
        self.assertEqual('Klarvær', data['tomorrow']['description'])
        self.assertEqual('nordvest', data['tomorrow']['wind']['direction'])
        self.assertEqual(2.2, data['tomorrow']['wind']['speed'])


class HackerNews(unittest.TestCase):

    def setUp(self):
        self.html = test_data('hn.html')
        self.hn = hackernews.HackerNews({'interval': None})

    def test_parse(self):
        data = self.hn._parse(self.html)

        self.assertEqual(30, len(data['items']))
        self.assertTrue('points' in data['items'][3])
        self.assertTrue('title' in data['items'][3])
        self.assertEqual(76, data['items'][3]['points'])
        self.assertEqual('Building an OpenBSD Router',
{'fraction_non_alphanumeric': 0.14985938127762152, 'fraction_numerical': 0.016070711128967456, 'mean_word_length': 2.657361151924772, 'pattern_counts': {'":': 12, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5782656', 'n_tokens_mistral': 3862, 'n_tokens_neox': 3667, 'n_words': 729}