TestWebUI.test_h005_rate_wrong_format() - Code Metrics - Inspection of "fix pause not reset properly, increase test covera..." - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f2ad1f...2fd0d7 )

by Roy

created 2016-08-30 00:12 UTC

TestWebUI.test_h005_rate_wrong_format() A

↳ Parent: TestWebUI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	1
c	1
b	1
f	0
dl	0
loc	7
rs	9.4285

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
# Author: Binux<[email protected]>
#         http://binux.me
# Created on 2014-11-18 21:03:22

import os
import re
import time
import json
import shutil
import unittest2 as unittest

from pyspider import run
from pyspider.libs import utils
from pyspider.libs.utils import run_in_thread, ObjectDict


class TestWebUI(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        shutil.rmtree('./data/tests', ignore_errors=True)
        os.makedirs('./data/tests')

        import tests.data_test_webpage
        import httpbin
        from pyspider.webui import bench_test  # flake8: noqa
        self.httpbin_thread = utils.run_in_subprocess(httpbin.app.run, port=14887, passthrough_errors=False)
        self.httpbin = 'http://127.0.0.1:14887'

        ctx = run.cli.make_context('test', [
            '--taskdb', 'sqlalchemy+sqlite+taskdb:///data/tests/task.db',
            '--projectdb', 'sqlalchemy+sqlite+projectdb:///data/tests/projectdb.db',
            '--resultdb', 'sqlalchemy+sqlite+resultdb:///data/tests/resultdb.db',
        ], None, obj=ObjectDict(testing_mode=True))
        self.ctx = run.cli.invoke(ctx)

        self.threads = []

        ctx = run.scheduler.make_context('scheduler', [], self.ctx)
        self.scheduler = scheduler = run.scheduler.invoke(ctx)
        self.threads.append(run_in_thread(scheduler.xmlrpc_run))
        self.threads.append(run_in_thread(scheduler.run))

        ctx = run.fetcher.make_context('fetcher', [
            '--xmlrpc',
            '--xmlrpc-port', '24444',
        ], self.ctx)
        fetcher = run.fetcher.invoke(ctx)
        self.threads.append(run_in_thread(fetcher.xmlrpc_run))
        self.threads.append(run_in_thread(fetcher.run))

        ctx = run.processor.make_context('processor', [], self.ctx)
        processor = run.processor.invoke(ctx)
        self.threads.append(run_in_thread(processor.run))

        ctx = run.result_worker.make_context('result_worker', [], self.ctx)
        result_worker = run.result_worker.invoke(ctx)
        self.threads.append(run_in_thread(result_worker.run))

        ctx = run.webui.make_context('webui', [
            '--scheduler-rpc', 'http://localhost:23333/'
        ], self.ctx)
        app = run.webui.invoke(ctx)
        app.debug = True
        self.app = app.test_client()
        self.rpc = app.config['scheduler_rpc']

        time.sleep(1)


    @classmethod
    def tearDownClass(self):
        for each in self.ctx.obj.instances:
            each.quit()
        time.sleep(1)

        for thread in self.threads:
            thread.join()

        self.httpbin_thread.terminate()
        self.httpbin_thread.join()

        assert not utils.check_port_open(5000)
        assert not utils.check_port_open(23333)
        assert not utils.check_port_open(24444)
        assert not utils.check_port_open(25555)
        assert not utils.check_port_open(14887)

        shutil.rmtree('./data/tests', ignore_errors=True)

    def test_10_index_page(self):
        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'dashboard', rv.data)

    def test_20_debug(self):
        rv = self.app.get('/debug/test_project')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'debugger', rv.data)
        self.assertIn(b'var task_content = ', rv.data)
        self.assertIn(b'var script_content = ', rv.data)

        m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.assertIn('test_project', json.loads(m.group(1)))

        m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.assertIn('__START_URL__', json.loads(m.group(1)))


    def test_25_debug_post(self):
        rv = self.app.post('/debug/test_project', data={
            'project-name': 'other_project',
            'start-urls': 'http://127.0.0.1:14887/pyspider/test.html',
            'script-mode': 'script',
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'debugger', rv.data)
        self.assertIn(b'var task_content = ', rv.data)
        self.assertIn(b'var script_content = ', rv.data)

        m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.assertIn('test_project', m.group(1))
        self.__class__.task_content = json.loads(m.group(1))

        m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.assertIn('127.0.0.1:14887', m.group(1))
        self.__class__.script_content = json.loads(m.group(1))


    def test_30_run(self):
        rv = self.app.post('/debug/test_project/run', data={
            'script': self.script_content,
            'task': self.task_content
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertIn(b'follows', rv.data)
        self.assertGreater(len(data['follows']), 0)
        self.__class__.task_content2 = data['follows'][0]

    def test_32_run_bad_task(self):
        rv = self.app.post('/debug/test_project/run', data={
            'script': self.script_content,
            'task': self.task_content+'asdfasdf312!@#'
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertGreater(len(data['logs']), 0)
        self.assertEqual(len(data['follows']), 0)

    def test_33_run_bad_script(self):
        rv = self.app.post('/debug/test_project/run', data={
            'script': self.script_content+'adfasfasdf',
            'task': self.task_content
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertGreater(len(data['logs']), 0)
        self.assertEqual(len(data['follows']), 0)

    def test_35_run_http_task(self):
        rv = self.app.post('/debug/test_project/run', data={
            'script': self.script_content,
            'task': json.dumps(self.task_content2)
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertIn('follows', data)

    def test_40_save(self):
        rv = self.app.post('/debug/test_project/save', data={
            'script': self.script_content,
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

    def test_42_get(self):
        rv = self.app.get('/debug/test_project/get')
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertIn('script', data)
        self.assertEqual(data['script'], self.script_content)


    def test_45_run_with_saved_script(self):
        rv = self.app.post('/debug/test_project/run', data={
            'webdav_mode': 'true',
            'script': '',
            'task': self.task_content
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertIn(b'follows', rv.data)
        self.assertGreater(len(data['follows']), 0)
        self.__class__.task_content2 = data['follows'][0]

    def test_50_index_page_list(self):
        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'"test_project"', rv.data)

    def test_52_change_status(self):
        rv = self.app.post('/update', data={
            'name': 'status',
            'value': 'RUNNING',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

    def test_55_reopen(self):
        rv = self.app.get('/debug/test_project')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'debugger', rv.data)

    def test_57_resave(self):
        rv = self.app.post('/debug/test_project/save', data={
            'script': self.script_content,
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

    def test_58_index_page_list(self):
        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'CHECKING', rv.data)

    def test_60_change_rate(self):
        rv = self.app.post('/update', data={
            'name': 'rate',
            'value': '1/4',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

    def test_70_change_status(self):
        rv = self.app.post('/update', data={
            'name': 'status',
            'value': 'RUNNING',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

    def test_80_change_group(self):
        rv = self.app.post('/update', data={
            'name': 'group',
            'value': 'test_binux',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'test_binux', rv.data)

    def test_90_run(self):
        time.sleep(0.5)
        rv = self.app.post('/run', data={
            'project': 'test_project',
        })
        self.assertEqual(rv.status_code, 200)
        self.assertEqual(json.loads(utils.text(rv.data))['result'], True)

    def test_a10_counter(self):
        for i in range(30):
            time.sleep(1)
            if self.rpc.counter('5m', 'sum')\
                    .get('test_project', {}).get('success', 0) > 5:
                break

        rv = self.app.get('/counter')
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertGreater(len(data), 0)
        self.assertGreater(data['test_project']['5m']['success'], 3)
        self.assertGreater(data['test_project']['1h']['success'], 3)
        self.assertGreater(data['test_project']['1d']['success'], 3)
        self.assertGreater(data['test_project']['all']['success'], 3)


    def test_a15_queues(self):
        rv = self.app.get('/queues')
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertGreater(len(data), 0)
        self.assertIn('scheduler2fetcher', data)
        self.assertIn('fetcher2processor', data)
        self.assertIn('processor2result', data)
        self.assertIn('newtask_queue', data)
        self.assertIn('status_queue', data)

    def test_a20_tasks(self):
        rv = self.app.get('/tasks')
        self.assertEqual(rv.status_code, 200, rv.data)
        self.assertIn(b'SUCCESS</span>', rv.data)
        self.assertNotIn(b'>ERROR</span>', rv.data)
        m = re.search(r'/task/test_project:[^"]+', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.__class__.task_url = m.group(0)
        self.assertIsNotNone(self.task_url)
        m = re.search(r'/debug/test_project[^"]+', utils.text(rv.data))
        self.assertIsNotNone(m)
        self.__class__.debug_task_url = m.group(0)
        self.assertIsNotNone(self.debug_task_url)

        rv = self.app.get('/tasks?project=test_project')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'SUCCESS</span>', rv.data)
        self.assertNotIn(b'>ERROR</span>', rv.data)

    def test_a22_active_tasks(self):
        rv = self.app.get('/active_tasks')
        data = json.loads(utils.text(rv.data))
        track = False
        self.assertGreater(len(data), 0)
        for task in data:
            for k in ('taskid', 'project', 'url', 'updatetime'):
                self.assertIn(k, task)
            if task.get('track'):
                track = True
                self.assertIn('fetch', task['track'])
                self.assertIn('ok', task['track']['fetch'])
                self.assertIn('time', task['track']['fetch'])
                self.assertIn('process', task['track'])
                self.assertIn('ok', task['track']['process'])
                self.assertIn('time', task['track']['process'])
        self.assertTrue(track)
                    

    def test_a24_task(self):
        rv = self.app.get(self.task_url)
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'lastcrawltime', rv.data)

    def test_a26_debug_task(self):
        rv = self.app.get(self.debug_task_url)
        self.assertEqual(rv.status_code, 200)

    def test_a30_results(self):
        rv = self.app.get('/results?project=test_project')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'<th>url</th>', rv.data)
        self.assertIn(b'open-url', rv.data)

    def test_a30_export_json(self):
        rv = self.app.get('/results/dump/test_project.json')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'"taskid":', rv.data)

    def test_a32_export_json_style_full(self):
        rv = self.app.get('/results/dump/test_project.json?style=full')
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode('utf8'))
        self.assertGreater(len(data), 1)

    def test_a34_export_json_style_full_limit_1(self):
        rv = self.app.get('/results/dump/test_project.json?style=full&limit=1&offset=1')
        self.assertEqual(rv.status_code, 200)
        data = json.loads(rv.data.decode('utf8'))
        self.assertEqual(len(data), 1)


    def test_a40_export_url_json(self):
        rv = self.app.get('/results/dump/test_project.txt')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'"url":', rv.data)

    def test_a50_export_csv(self):
        rv = self.app.get('/results/dump/test_project.csv')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'url,title,url', rv.data)

    def test_a60_fetch_via_cannot_connect_fetcher(self):
        ctx = run.webui.make_context('webui', [
            '--fetcher-rpc', 'http://localhost:20000/',
        ], self.ctx)

        app = run.webui.invoke(ctx)
        app = app.test_client()
        rv = app.post('/debug/test_project/run', data={
            'script': self.script_content,
            'task': self.task_content
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertGreater(len(data['logs']), 0)
        self.assertEqual(len(data['follows']), 0)

    def test_a70_fetch_via_fetcher(self):
        ctx = run.webui.make_context('webui', [
            '--fetcher-rpc', 'http://localhost:24444/',
        ], self.ctx)
        app = run.webui.invoke(ctx)
        app = app.test_client()
        rv = app.post('/debug/test_project/run', data={
            'script': self.script_content,
            'task': self.task_content
        })
        self.assertEqual(rv.status_code, 200)
        data = json.loads(utils.text(rv.data))
        self.assertEqual(len(data['logs']), 0, data['logs'])
        self.assertIn(b'follows', rv.data)
        self.assertGreater(len(data['follows']), 0)

    def test_h000_auth(self):
        ctx = run.webui.make_context('webui', [
            '--scheduler-rpc', 'http://localhost:23333/',
            '--username', 'binux',
            '--password', '4321',
        ], self.ctx)
        app = run.webui.invoke(ctx)
        self.__class__.app = app.test_client()
        self.__class__.rpc = app.config['scheduler_rpc']

    def test_h005_no_such_project(self):
        rv = self.app.post('/update', data={
            'name': 'group',
            'value': 'lock',
            'pk': 'not_exist_project'
        })
        self.assertEqual(rv.status_code, 404)

    def test_h005_unknown_field(self):
        rv = self.app.post('/update', data={
            'name': 'unknown_field',
            'value': 'lock',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 400)

    def test_h005_rate_wrong_format(self):
        rv = self.app.post('/update', data={
            'name': 'rate',
            'value': 'xxx',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 400)

    def test_h010_change_group(self):
        rv = self.app.post('/update', data={
            'name': 'group',
            'value': 'lock',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ok', rv.data)

        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'lock', rv.data)

    def test_h020_change_group_lock_failed(self):
        rv = self.app.post('/update', data={
            'name': 'group',
            'value': '',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 401)

    def test_h020_change_group_lock_ok(self):
        rv = self.app.post('/update', data={
            'name': 'group',
            'value': 'test_binux',
            'pk': 'test_project'
        }, headers={
            'Authorization': 'Basic YmludXg6NDMyMQ=='
        })
        self.assertEqual(rv.status_code, 200)

    def test_h030_need_auth(self):
        ctx = run.webui.make_context('webui', [
            '--scheduler-rpc', 'http://localhost:23333/',
            '--username', 'binux',
            '--password', '4321',
            '--need-auth',
        ], self.ctx)
        app = run.webui.invoke(ctx)
        self.__class__.app = app.test_client()
        self.__class__.rpc = app.config['scheduler_rpc']

    def test_h040_auth_fail(self):
        rv = self.app.get('/')
        self.assertEqual(rv.status_code, 401)

    def test_h050_auth_fail2(self):
        rv = self.app.get('/', headers={
            'Authorization': 'Basic Ymlasdfsd'
        })
        self.assertEqual(rv.status_code, 401)

    def test_h060_auth_fail3(self):
        rv = self.app.get('/', headers={
            'Authorization': 'Basic YmludXg6MQ=='
        })
        self.assertEqual(rv.status_code, 401)

    def test_h070_auth_ok(self):
        rv = self.app.get('/', headers={
            'Authorization': 'Basic YmludXg6NDMyMQ=='
        })
        self.assertEqual(rv.status_code, 200)

    def test_x0_disconnected_scheduler(self):
        ctx = run.webui.make_context('webui', [
            '--scheduler-rpc', 'http://localhost:23458/'
        ], self.ctx)
        app = run.webui.invoke(ctx)
        self.__class__.app = app.test_client()
        self.__class__.rpc = app.config['scheduler_rpc']

    def test_x10_project_update(self):
        rv = self.app.post('/update', data={
            'name': 'status',
            'value': 'RUNNING',
            'pk': 'test_project'
        })
        self.assertEqual(rv.status_code, 200)
        self.assertNotIn(b'ok', rv.data)

    def test_x20_counter(self):
        rv = self.app.get('/counter?time=5m&type=sum')
        self.assertEqual(rv.status_code, 200)
        self.assertEqual(json.loads(utils.text(rv.data)), {})

    def test_x30_run_not_exists_project(self):
        rv = self.app.post('/run', data={
            'project': 'not_exist_project',
        })
        self.assertEqual(rv.status_code, 404)

    def test_x30_run(self):
        rv = self.app.post('/run', data={
            'project': 'test_project',
        })
        self.assertEqual(rv.status_code, 200)
        self.assertEqual(json.loads(utils.text(rv.data))['result'], False)

    def test_x40_debug_save(self):
        rv = self.app.post('/debug/test_project/save', data={
            'script': self.script_content,
        })
        self.assertEqual(rv.status_code, 200)
        self.assertNotIn(b'ok', rv.data)

    def test_x50_tasks(self):
        rv = self.app.get('/tasks')
        self.assertEqual(rv.status_code, 502)

    def test_x60_robots(self):
        rv = self.app.get('/robots.txt')
        self.assertEqual(rv.status_code, 200)
        self.assertIn(b'ser-agent', rv.data)

    def test_x70_bench(self):
        rv = self.app.get('/bench?total=10&show=5')
        self.assertEqual(rv.status_code, 200)


1		#!/usr/bin/env python
2		# -- encoding: utf-8 --
3		# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
4		# Author: Binux<[email protected]>
5		# http://binux.me
6		# Created on 2014-11-18 21:03:22
7
8		import os
9		import re
10		import time
11		import json
12		import shutil
13		import unittest2 as unittest
14
15		from pyspider import run
16		from pyspider.libs import utils
17		from pyspider.libs.utils import run_in_thread, ObjectDict
18
19
20		class TestWebUI(unittest.TestCase):
21
22		@classmethod
23		def setUpClass(self):
24		shutil.rmtree('./data/tests', ignore_errors=True)
25		os.makedirs('./data/tests')
26
27		import tests.data_test_webpage
28		import httpbin
29		from pyspider.webui import bench_test # flake8: noqa
30		self.httpbin_thread = utils.run_in_subprocess(httpbin.app.run, port=14887, passthrough_errors=False)
31		self.httpbin = 'http://127.0.0.1:14887'
32
33		ctx = run.cli.make_context('test', [
34		'--taskdb', 'sqlalchemy+sqlite+taskdb:///data/tests/task.db',
35		'--projectdb', 'sqlalchemy+sqlite+projectdb:///data/tests/projectdb.db',
36		'--resultdb', 'sqlalchemy+sqlite+resultdb:///data/tests/resultdb.db',
37		], None, obj=ObjectDict(testing_mode=True))
38		self.ctx = run.cli.invoke(ctx)
39
40		self.threads = []
41
42		ctx = run.scheduler.make_context('scheduler', [], self.ctx)
43		self.scheduler = scheduler = run.scheduler.invoke(ctx)
44		self.threads.append(run_in_thread(scheduler.xmlrpc_run))
45		self.threads.append(run_in_thread(scheduler.run))
46
47		ctx = run.fetcher.make_context('fetcher', [
48		'--xmlrpc',
49		'--xmlrpc-port', '24444',
50		], self.ctx)
51		fetcher = run.fetcher.invoke(ctx)
52		self.threads.append(run_in_thread(fetcher.xmlrpc_run))
53		self.threads.append(run_in_thread(fetcher.run))
54
55		ctx = run.processor.make_context('processor', [], self.ctx)
56		processor = run.processor.invoke(ctx)
57		self.threads.append(run_in_thread(processor.run))
58
59		ctx = run.result_worker.make_context('result_worker', [], self.ctx)
60		result_worker = run.result_worker.invoke(ctx)
61		self.threads.append(run_in_thread(result_worker.run))
62
63		ctx = run.webui.make_context('webui', [
64		'--scheduler-rpc', 'http://localhost:23333/'
65		], self.ctx)
66		app = run.webui.invoke(ctx)
67		app.debug = True
68		self.app = app.test_client()
69		self.rpc = app.config['scheduler_rpc']
70
71		time.sleep(1)
72	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
73		@classmethod
74		def tearDownClass(self):
75		for each in self.ctx.obj.instances:
76		each.quit()
77		time.sleep(1)
78
79		for thread in self.threads:
80		thread.join()
81
82		self.httpbin_thread.terminate()
83		self.httpbin_thread.join()
84
85		assert not utils.check_port_open(5000)
86		assert not utils.check_port_open(23333)
87		assert not utils.check_port_open(24444)
88		assert not utils.check_port_open(25555)
89		assert not utils.check_port_open(14887)
90
91		shutil.rmtree('./data/tests', ignore_errors=True)
92
93		def test_10_index_page(self):
94		rv = self.app.get('/')
95		self.assertEqual(rv.status_code, 200)
96		self.assertIn(b'dashboard', rv.data)
97
98		def test_20_debug(self):
99		rv = self.app.get('/debug/test_project')
100		self.assertEqual(rv.status_code, 200)
101		self.assertIn(b'debugger', rv.data)
102		self.assertIn(b'var task_content = ', rv.data)
103		self.assertIn(b'var script_content = ', rv.data)
104
105		m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
106		self.assertIsNotNone(m)
107		self.assertIn('test_project', json.loads(m.group(1)))
108
109		m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
110		self.assertIsNotNone(m)
111		self.assertIn('__START_URL__', json.loads(m.group(1)))
112	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
113		def test_25_debug_post(self):
114		rv = self.app.post('/debug/test_project', data={
115		'project-name': 'other_project',
116		'start-urls': 'http://127.0.0.1:14887/pyspider/test.html',
117		'script-mode': 'script',
118		})
119		self.assertEqual(rv.status_code, 200)
120		self.assertIn(b'debugger', rv.data)
121		self.assertIn(b'var task_content = ', rv.data)
122		self.assertIn(b'var script_content = ', rv.data)
123
124		m = re.search(r'var task_content = (.*);\n', utils.text(rv.data))
125		self.assertIsNotNone(m)
126		self.assertIn('test_project', m.group(1))
127		self.__class__.task_content = json.loads(m.group(1))
128
129		m = re.search(r'var script_content = (.*);\n', utils.text(rv.data))
130		self.assertIsNotNone(m)
131		self.assertIn('127.0.0.1:14887', m.group(1))
132		self.__class__.script_content = json.loads(m.group(1))
133	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
134		def test_30_run(self):
135		rv = self.app.post('/debug/test_project/run', data={
136		'script': self.script_content,
137		'task': self.task_content
138		})
139		self.assertEqual(rv.status_code, 200)
140		data = json.loads(utils.text(rv.data))
141		self.assertIn(b'follows', rv.data)
142		self.assertGreater(len(data['follows']), 0)
143		self.__class__.task_content2 = data['follows'][0]
144
145		def test_32_run_bad_task(self):
146		rv = self.app.post('/debug/test_project/run', data={
147		'script': self.script_content,
148		'task': self.task_content+'asdfasdf312!@#'
149		})
150		self.assertEqual(rv.status_code, 200)
151		data = json.loads(utils.text(rv.data))
152		self.assertGreater(len(data['logs']), 0)
153		self.assertEqual(len(data['follows']), 0)
154
155		def test_33_run_bad_script(self):
156		rv = self.app.post('/debug/test_project/run', data={
157		'script': self.script_content+'adfasfasdf',
158		'task': self.task_content
159		})
160		self.assertEqual(rv.status_code, 200)
161		data = json.loads(utils.text(rv.data))
162		self.assertGreater(len(data['logs']), 0)
163		self.assertEqual(len(data['follows']), 0)
164
165		def test_35_run_http_task(self):
166		rv = self.app.post('/debug/test_project/run', data={
167		'script': self.script_content,
168		'task': json.dumps(self.task_content2)
169		})
170		self.assertEqual(rv.status_code, 200)
171		data = json.loads(utils.text(rv.data))
172		self.assertIn('follows', data)
173
174		def test_40_save(self):
175		rv = self.app.post('/debug/test_project/save', data={
176		'script': self.script_content,
177		})
178		self.assertEqual(rv.status_code, 200)
179		self.assertIn(b'ok', rv.data)
180
181		def test_42_get(self):
182		rv = self.app.get('/debug/test_project/get')
183		self.assertEqual(rv.status_code, 200)
184		data = json.loads(utils.text(rv.data))
185		self.assertIn('script', data)
186		self.assertEqual(data['script'], self.script_content)
187	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
188		def test_45_run_with_saved_script(self):
189		rv = self.app.post('/debug/test_project/run', data={
190		'webdav_mode': 'true',
191		'script': '',
192		'task': self.task_content
193		})
194		self.assertEqual(rv.status_code, 200)
195		data = json.loads(utils.text(rv.data))
196		self.assertIn(b'follows', rv.data)
197		self.assertGreater(len(data['follows']), 0)
198		self.__class__.task_content2 = data['follows'][0]
199
200		def test_50_index_page_list(self):
201		rv = self.app.get('/')
202		self.assertEqual(rv.status_code, 200)
203		self.assertIn(b'"test_project"', rv.data)
204
205		def test_52_change_status(self):
206		rv = self.app.post('/update', data={
207		'name': 'status',
208		'value': 'RUNNING',
209		'pk': 'test_project'
210		})
211		self.assertEqual(rv.status_code, 200)
212		self.assertIn(b'ok', rv.data)
213
214		def test_55_reopen(self):
215		rv = self.app.get('/debug/test_project')
216		self.assertEqual(rv.status_code, 200)
217		self.assertIn(b'debugger', rv.data)
218
219		def test_57_resave(self):
220		rv = self.app.post('/debug/test_project/save', data={
221		'script': self.script_content,
222		})
223		self.assertEqual(rv.status_code, 200)
224		self.assertIn(b'ok', rv.data)
225
226		def test_58_index_page_list(self):
227		rv = self.app.get('/')
228		self.assertEqual(rv.status_code, 200)
229		self.assertIn(b'CHECKING', rv.data)
230
231		def test_60_change_rate(self):
232		rv = self.app.post('/update', data={
233		'name': 'rate',
234		'value': '1/4',
235		'pk': 'test_project'
236		})
237		self.assertEqual(rv.status_code, 200)
238		self.assertIn(b'ok', rv.data)
239
240		def test_70_change_status(self):
241		rv = self.app.post('/update', data={
242		'name': 'status',
243		'value': 'RUNNING',
244		'pk': 'test_project'
245		})
246		self.assertEqual(rv.status_code, 200)
247		self.assertIn(b'ok', rv.data)
248
249		def test_80_change_group(self):
250		rv = self.app.post('/update', data={
251		'name': 'group',
252		'value': 'test_binux',
253		'pk': 'test_project'
254		})
255		self.assertEqual(rv.status_code, 200)
256		self.assertIn(b'ok', rv.data)
257
258		rv = self.app.get('/')
259		self.assertEqual(rv.status_code, 200)
260		self.assertIn(b'test_binux', rv.data)
261
262		def test_90_run(self):
263		time.sleep(0.5)
264		rv = self.app.post('/run', data={
265		'project': 'test_project',
266		})
267		self.assertEqual(rv.status_code, 200)
268		self.assertEqual(json.loads(utils.text(rv.data))['result'], True)
269
270		def test_a10_counter(self):
271		for i in range(30):
272		time.sleep(1)
273		if self.rpc.counter('5m', 'sum')\
274		.get('test_project', {}).get('success', 0) > 5:
275		break
276
277		rv = self.app.get('/counter')
278		self.assertEqual(rv.status_code, 200)
279		data = json.loads(utils.text(rv.data))
280		self.assertGreater(len(data), 0)
281		self.assertGreater(data['test_project']['5m']['success'], 3)
282		self.assertGreater(data['test_project']['1h']['success'], 3)
283		self.assertGreater(data['test_project']['1d']['success'], 3)
284		self.assertGreater(data['test_project']['all']['success'], 3)
285	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
286		def test_a15_queues(self):
287		rv = self.app.get('/queues')
288		self.assertEqual(rv.status_code, 200)
289		data = json.loads(utils.text(rv.data))
290		self.assertGreater(len(data), 0)
291		self.assertIn('scheduler2fetcher', data)
292		self.assertIn('fetcher2processor', data)
293		self.assertIn('processor2result', data)
294		self.assertIn('newtask_queue', data)
295		self.assertIn('status_queue', data)
296
297		def test_a20_tasks(self):
298		rv = self.app.get('/tasks')
299		self.assertEqual(rv.status_code, 200, rv.data)
300		self.assertIn(b'SUCCESS</span>', rv.data)
301		self.assertNotIn(b'>ERROR</span>', rv.data)
302		m = re.search(r'/task/test_project:[^"]+', utils.text(rv.data))
303		self.assertIsNotNone(m)
304		self.__class__.task_url = m.group(0)
305		self.assertIsNotNone(self.task_url)
306		m = re.search(r'/debug/test_project[^"]+', utils.text(rv.data))
307		self.assertIsNotNone(m)
308		self.__class__.debug_task_url = m.group(0)
309		self.assertIsNotNone(self.debug_task_url)
310
311		rv = self.app.get('/tasks?project=test_project')
312		self.assertEqual(rv.status_code, 200)
313		self.assertIn(b'SUCCESS</span>', rv.data)
314		self.assertNotIn(b'>ERROR</span>', rv.data)
315
316		def test_a22_active_tasks(self):
317		rv = self.app.get('/active_tasks')
318		data = json.loads(utils.text(rv.data))
319		track = False
320		self.assertGreater(len(data), 0)
321		for task in data:
322		for k in ('taskid', 'project', 'url', 'updatetime'):
323		self.assertIn(k, task)
324		if task.get('track'):
325		track = True
326		self.assertIn('fetch', task['track'])
327		self.assertIn('ok', task['track']['fetch'])
328		self.assertIn('time', task['track']['fetch'])
329		self.assertIn('process', task['track'])
330		self.assertIn('ok', task['track']['process'])
331		self.assertIn('time', task['track']['process'])
332		self.assertTrue(track)
333
334
335		def test_a24_task(self):
336		rv = self.app.get(self.task_url)
337		self.assertEqual(rv.status_code, 200)
338		self.assertIn(b'lastcrawltime', rv.data)
339
340		def test_a26_debug_task(self):
341		rv = self.app.get(self.debug_task_url)
342		self.assertEqual(rv.status_code, 200)
343
344		def test_a30_results(self):
345		rv = self.app.get('/results?project=test_project')
346		self.assertEqual(rv.status_code, 200)
347		self.assertIn(b'<th>url</th>', rv.data)
348		self.assertIn(b'open-url', rv.data)
349
350		def test_a30_export_json(self):
351		rv = self.app.get('/results/dump/test_project.json')
352		self.assertEqual(rv.status_code, 200)
353		self.assertIn(b'"taskid":', rv.data)
354
355		def test_a32_export_json_style_full(self):
356		rv = self.app.get('/results/dump/test_project.json?style=full')
357		self.assertEqual(rv.status_code, 200)
358		data = json.loads(rv.data.decode('utf8'))
359		self.assertGreater(len(data), 1)
360
361		def test_a34_export_json_style_full_limit_1(self):
362		rv = self.app.get('/results/dump/test_project.json?style=full&limit=1&offset=1')
363		self.assertEqual(rv.status_code, 200)
364		data = json.loads(rv.data.decode('utf8'))
365	View Code Duplication	self.assertEqual(len(data), 1)
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
366
367		def test_a40_export_url_json(self):
368		rv = self.app.get('/results/dump/test_project.txt')
369		self.assertEqual(rv.status_code, 200)
370		self.assertIn(b'"url":', rv.data)
371
372		def test_a50_export_csv(self):
373		rv = self.app.get('/results/dump/test_project.csv')
374		self.assertEqual(rv.status_code, 200)
375		self.assertIn(b'url,title,url', rv.data)
376
377		def test_a60_fetch_via_cannot_connect_fetcher(self):
378		ctx = run.webui.make_context('webui', [
379		'--fetcher-rpc', 'http://localhost:20000/',
380	View Code Duplication	], self.ctx)
		0 ignored issues – show Duplication introduced 2016-04-26 19:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
381		app = run.webui.invoke(ctx)
382		app = app.test_client()
383		rv = app.post('/debug/test_project/run', data={
384		'script': self.script_content,
385		'task': self.task_content
386		})
387		self.assertEqual(rv.status_code, 200)
388		data = json.loads(utils.text(rv.data))
389		self.assertGreater(len(data['logs']), 0)
390		self.assertEqual(len(data['follows']), 0)
391
392		def test_a70_fetch_via_fetcher(self):
393		ctx = run.webui.make_context('webui', [
394		'--fetcher-rpc', 'http://localhost:24444/',
395		], self.ctx)
396		app = run.webui.invoke(ctx)
397		app = app.test_client()
398		rv = app.post('/debug/test_project/run', data={
399		'script': self.script_content,
400		'task': self.task_content
401		})
402		self.assertEqual(rv.status_code, 200)
403		data = json.loads(utils.text(rv.data))
404		self.assertEqual(len(data['logs']), 0, data['logs'])
405		self.assertIn(b'follows', rv.data)
406		self.assertGreater(len(data['follows']), 0)
407
408		def test_h000_auth(self):
409		ctx = run.webui.make_context('webui', [
410		'--scheduler-rpc', 'http://localhost:23333/',
411		'--username', 'binux',
412		'--password', '4321',
413		], self.ctx)
414		app = run.webui.invoke(ctx)
415		self.__class__.app = app.test_client()
416		self.__class__.rpc = app.config['scheduler_rpc']
417
418		def test_h005_no_such_project(self):
419		rv = self.app.post('/update', data={
420		'name': 'group',
421		'value': 'lock',
422		'pk': 'not_exist_project'
423		})
424		self.assertEqual(rv.status_code, 404)
425
426		def test_h005_unknown_field(self):
427		rv = self.app.post('/update', data={
428		'name': 'unknown_field',
429		'value': 'lock',
430		'pk': 'test_project'
431		})
432		self.assertEqual(rv.status_code, 400)
433
434		def test_h005_rate_wrong_format(self):
435		rv = self.app.post('/update', data={
436		'name': 'rate',
437		'value': 'xxx',
438		'pk': 'test_project'
439		})
440		self.assertEqual(rv.status_code, 400)
441
442		def test_h010_change_group(self):
443		rv = self.app.post('/update', data={
444		'name': 'group',
445		'value': 'lock',
446		'pk': 'test_project'
447		})
448		self.assertEqual(rv.status_code, 200)
449		self.assertIn(b'ok', rv.data)
450
451		rv = self.app.get('/')
452		self.assertEqual(rv.status_code, 200)
453		self.assertIn(b'lock', rv.data)
454
455		def test_h020_change_group_lock_failed(self):
456		rv = self.app.post('/update', data={
457		'name': 'group',
458		'value': '',
459		'pk': 'test_project'
460		})
461		self.assertEqual(rv.status_code, 401)
462
463		def test_h020_change_group_lock_ok(self):
464		rv = self.app.post('/update', data={
465		'name': 'group',
466		'value': 'test_binux',
467		'pk': 'test_project'
468		}, headers={
469		'Authorization': 'Basic YmludXg6NDMyMQ=='
470		})
471		self.assertEqual(rv.status_code, 200)
472
473		def test_h030_need_auth(self):
474		ctx = run.webui.make_context('webui', [
475		'--scheduler-rpc', 'http://localhost:23333/',
476		'--username', 'binux',
477		'--password', '4321',
478		'--need-auth',
479		], self.ctx)
480		app = run.webui.invoke(ctx)
481		self.__class__.app = app.test_client()
482		self.__class__.rpc = app.config['scheduler_rpc']
483
484		def test_h040_auth_fail(self):
485		rv = self.app.get('/')
486		self.assertEqual(rv.status_code, 401)
487
488		def test_h050_auth_fail2(self):
489		rv = self.app.get('/', headers={
490		'Authorization': 'Basic Ymlasdfsd'
491		})
492		self.assertEqual(rv.status_code, 401)
493
494		def test_h060_auth_fail3(self):
495		rv = self.app.get('/', headers={
496		'Authorization': 'Basic YmludXg6MQ=='
497		})
498		self.assertEqual(rv.status_code, 401)
499
500		def test_h070_auth_ok(self):
501		rv = self.app.get('/', headers={
502		'Authorization': 'Basic YmludXg6NDMyMQ=='
503		})
504		self.assertEqual(rv.status_code, 200)
505
506		def test_x0_disconnected_scheduler(self):
507		ctx = run.webui.make_context('webui', [
508		'--scheduler-rpc', 'http://localhost:23458/'
509		], self.ctx)
510		app = run.webui.invoke(ctx)
511		self.__class__.app = app.test_client()
512		self.__class__.rpc = app.config['scheduler_rpc']
513
514		def test_x10_project_update(self):
515		rv = self.app.post('/update', data={
516		'name': 'status',
517		'value': 'RUNNING',
518		'pk': 'test_project'
519		})
520		self.assertEqual(rv.status_code, 200)
521		self.assertNotIn(b'ok', rv.data)
522
523		def test_x20_counter(self):
524		rv = self.app.get('/counter?time=5m&type=sum')
525		self.assertEqual(rv.status_code, 200)
526		self.assertEqual(json.loads(utils.text(rv.data)), {})
527
528		def test_x30_run_not_exists_project(self):
529		rv = self.app.post('/run', data={
530		'project': 'not_exist_project',
531		})
532		self.assertEqual(rv.status_code, 404)
533
534		def test_x30_run(self):
535		rv = self.app.post('/run', data={
536		'project': 'test_project',
537		})
538		self.assertEqual(rv.status_code, 200)
539		self.assertEqual(json.loads(utils.text(rv.data))['result'], False)
540
541		def test_x40_debug_save(self):
542		rv = self.app.post('/debug/test_project/save', data={
543		'script': self.script_content,
544		})
545		self.assertEqual(rv.status_code, 200)
546		self.assertNotIn(b'ok', rv.data)
547
548		def test_x50_tasks(self):
549		rv = self.app.get('/tasks')
550		self.assertEqual(rv.status_code, 502)
551
552		def test_x60_robots(self):
553		rv = self.app.get('/robots.txt')
554		self.assertEqual(rv.status_code, 200)
555		self.assertIn(b'ser-agent', rv.data)
556
557		def test_x70_bench(self):
558		rv = self.app.get('/bench?total=10&show=5')
559		self.assertEqual(rv.status_code, 200)
560

binux / pyspider

Push — master ( f2ad1f...2fd0d7 )

TestWebUI.test_h005_rate_wrong_format() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like