File: test_base.py

package info (click to toggle)
python-logi-circle 0.2.3-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 300 kB
  • sloc: python: 1,685; xml: 16; sh: 5; makefile: 4
file content (68 lines) | stat: -rw-r--r-- 2,309 bytes parent folder | download
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
"""Register Logi API with mock aiohttp ClientSession and responses."""
import os
import pickle
import json
import unittest
import asyncio
from tests.helpers import get_fixtures
from logi_circle.const import DEFAULT_SCOPES
from logi_circle.auth import AuthProvider

CLIENT_ID = 'abcdefghijklmnopqrstuvwxyz'
CLIENT_SECRET = 'correct_horse_battery_staple'
REDIRECT_URI = 'https://my.groovy.app/'
API_KEY = 'ZYXWVUTSRQPONMLKJIHGFEDCBA'
CACHE_FILE = os.path.join(os.path.dirname(__file__), 'cache.db')
FIXTURES = get_fixtures()


class LogiUnitTestBase(unittest.TestCase):
    """Base Logi Circle test class."""

    def setUp(self):
        """Setup unit test, create event loop."""
        from logi_circle import LogiCircle

        self.logi = LogiCircle(client_id=CLIENT_ID,
                               client_secret=CLIENT_SECRET,
                               redirect_uri=REDIRECT_URI,
                               cache_file=CACHE_FILE,
                               api_key=API_KEY)
        self.fixtures = FIXTURES
        self.client_id = CLIENT_ID
        self.client_secret = CLIENT_SECRET
        self.redirect_uri = REDIRECT_URI
        self.cache_file = CACHE_FILE

        self.loop = asyncio.new_event_loop()

    def get_authorized_auth_provider(self):
        """Returns a pre-authorized AuthProvider instance"""
        auth_fixture = json.loads(self.fixtures['auth_code'])
        token = {}
        token[self.client_id] = auth_fixture
        with open(self.cache_file, 'wb') as pickle_db:
            pickle.dump(token, pickle_db)

        return AuthProvider(client_id=self.client_id,
                            client_secret=self.client_secret,
                            redirect_uri=self.redirect_uri,
                            scopes=DEFAULT_SCOPES,
                            cache_file=self.cache_file,
                            logi_base=self.logi)

    def cleanup(self):
        """Cleanup any data created from the tests."""
        async def close_session():
            await self.logi.close()

        self.loop.run_until_complete(close_session())

        self.loop.close()
        self.logi = None
        if os.path.isfile(CACHE_FILE):
            os.remove(CACHE_FILE)

    def tearDown(self):
        """Stop everything started."""
        self.cleanup()