File: tests_update.py

package info (click to toggle)
paperwork 2.2.5-3
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 166,660 kB
  • sloc: python: 44,775; makefile: 992; sh: 625; xml: 135
file content (107 lines) | stat: -rw-r--r-- 3,623 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
import datetime
import http
import http.server
import json
import unittest
import threading
import time

import openpaperwork_core


class TestUpdate(unittest.TestCase):
    def setUp(self):
        self.core = openpaperwork_core.Core(auto_load_dependencies=True)
        self.core.load("openpaperwork_core.config.fake")
        self.core.load("paperwork_backend.app")
        self.core.load("paperwork_backend.beacon.update")

        self.config = self.core.get_by_name("openpaperwork_core.config.fake")
        self.received = False

    def test_check_update(self):
        self.received = False

        class TestRequestHandler(http.server.BaseHTTPRequestHandler):
            def do_GET(s):
                self.assertEqual(s.path, "/beacon/latest")
                s.send_response(200)
                s.send_header('Content-type', 'application/json')
                s.end_headers()
                s.wfile.write(json.dumps({
                    "paperwork": {
                        "posix": "999.1.2",
                        "nt": "999.1.2",
                    },
                }).encode("utf-8"))
                self.received = True

        with http.server.HTTPServer(('', 0), TestRequestHandler) as h:
            self.config.settings = {
                "check_for_update": True,
                "last_update_found": "0.1.3",
                "update_last_run": datetime.date(1995, 1, 1),
                "update_protocol": "http",
                "update_server": "127.0.0.1:{}".format(h.server_port),
            }

            threading.Thread(target=h.handle_request).start()
            time.sleep(0.1)

            class FakeModule(object):
                class Plugin(openpaperwork_core.PluginBase):
                    def on_update_detected(s, current, new):
                        self.assertEqual(new, (999, 1, 2))
                        self.core.call_all("mainloop_quit_graceful")

            self.core._load_module(
                "fake_module", FakeModule()
            )

            self.core.init()

            self.core.call_one('mainloop')

        self.assertTrue(self.received)

    def test_server_down(self):
        self.received = False
        self.update_notified = False

        class TestRequestHandler(http.server.BaseHTTPRequestHandler):
            def do_GET(s):
                self.assertEqual(s.path, "/beacon/latest")
                s.send_response(500)
                s.send_header('Content-type', 'text/html')
                s.end_headers()
                s.wfile.write(b"<html><body><p>KO</p></body></html>")
                self.received = True

        with http.server.HTTPServer(('', 0), TestRequestHandler) as h:
            self.config.settings = {
                "check_for_update": True,
                "last_update_found": "0.1.3",
                "update_last_run": datetime.date(1995, 1, 1),
                "update_protocol": "http",
                "update_server": "127.0.0.1:{}".format(h.server_port),
            }

            threading.Thread(target=h.handle_request).start()
            time.sleep(0.1)

            class FakeModule(object):
                class Plugin(openpaperwork_core.PluginBase):
                    def on_update_detected(s, current, new):
                        self.update_notified = True

            self.core._load_module(
                "fake_module", FakeModule()
            )

            self.core.init()

            self.core.call_all("mainloop_quit_graceful")
            self.core.call_one('mainloop')

        self.assertTrue(self.received)
        self.assertFalse(self.update_notified)