wallabag-api-client/wallabag_api/wallabag_test.py

# coding: utf-8
import asyncio
import configparser
import datetime
import os
import sys
import uuid
import unittest

import aiohttp
from wallabag_api.wallabag import Wallabag

ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(ROOT_DIR, "testcfg.ini")


def load_test_config(path):
    config = configparser.ConfigParser()
    config.read(path)
    return config["Test Configuration"]


def run_async(f):
    def wrapper(*args, **kwargs):
        coro = asyncio.coroutine(f)
        future = coro(*args, **kwargs)
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(future)

    return wrapper


class TestWallabag(unittest.TestCase):

    entry_data = {
        "title": "foobar title",
        "url": "https://somwhere.over.the.raibow.com/",
        "tags": ["foo", "bar"],
        "starred": 0,
        "archive": 0,
        "content": "<p>Additional content</p>",
        "language": "FR",
        "published_at": datetime.datetime.now(),
        "authors": "John Doe",
        "public": 0,
        "original_url": "http://localhost",
    }

    @classmethod
    @run_async
    async def setUpClass(cls):
        cls._configuration = dict(load_test_config(CONFIG_PATH))
        cls.access_token = await Wallabag.get_token(**cls._configuration)
        # Open connection to wallabag
        cls.w = Wallabag(
            host=cls._configuration["host"],
            token=cls.access_token,
            client_id=cls._configuration["client_id"],
            client_secret=cls._configuration["client_secret"],
            aio_sess=aiohttp.ClientSession(),
        )
        # Create an entry for the tests
        cls.entry = await cls.w.post_entries(**cls.entry_data)

    @classmethod
    def tearDownClass(cls):
        asyncio.get_event_loop().run_until_complete(
            cls.w.delete_entries(cls.entry["id"])
        )
        asyncio.get_event_loop().run_until_complete(cls.w.aio_sess.close())

    @run_async
    async def test_get_token(self):
        data = await Wallabag.get_token(**self._configuration)
        print(data, file=sys.stderr)
        self.assertIsInstance(data, str)

    @run_async
    async def test_get_entries(self):
        data = await self.w.get_entries(
            **{
                "delete": 0,
                "sort": "created",
                "order": "desc",
                "page": 1,
                "perPage": 30,
                "tags": [],
            }
        )
        self.assertIsInstance(data, dict)

    @run_async
    async def test_get_entry(self):
        entry_id = self.entry["id"]
        self.assertIsInstance(entry_id, int)
        data = await self.w.get_entry(entry_id)
        self.assertIsInstance(data, dict)

    @run_async
    async def test_get_entry_tags(self):
        entry_id = self.entry["id"]
        self.assertIsInstance(entry_id, int)
        data = await self.w.get_entry_tags(entry_id)
        tag_names = [t["label"] for t in data]
        self.assertIsInstance(data, list)
        self.assertEquals(self.entry_data["tags"], tag_names)

    @run_async
    async def test_get_tags(self):
        data = await self.w.get_tags()
        self.assertIsInstance(data, list)

    @run_async
    async def test_post_entries(self):
        entry_data = self.entry_data.copy()
        entry_data["url"] = "https://example.org"
        data = await self.w.post_entries(**entry_data)
        self.assertTrue(data, True)
        self.assertIsInstance(data["id"], int)
        await self.w.delete_entries(data["id"])

    @run_async
    async def test_patch_entries(self):
        entry_id = self.entry["id"]
        self.assertIsInstance(entry_id, int)
        data = await self.w.patch_entries(
            entry_id,
            **{
                "title": "I change the title",
                "archive": 0,
                "tags": ["bimbo", "pipo"],
                "order": "asc",
                "star": 0,
                "delete": 0,
            },
        )
        self.assertTrue(data)

    @run_async
    async def test_delete_entries(self):
        entry_id = await self._create_entry()
        self.assertIsInstance(entry_id, int)
        data = await self.w.delete_entries(entry_id)
        self.assertTrue(data, True)

    @run_async
    async def test_post_entry_tags(self):
        entry_id = self.entry["id"]
        self.assertIsInstance(entry_id, int)
        tags = ["foo", "bar"]
        self.assertIsInstance(tags, list)
        data = await self.w.post_entry_tags(entry_id, tags)
        self.assertTrue(data, True)

    @run_async
    async def test_delete_entry_tag(self):
        entry_id = await self._create_entry()
        tag_id = 1
        self.assertIsInstance(entry_id, int)
        self.assertIsInstance(tag_id, int)
        resp = await self.w.delete_entry_tag(entry_id, tag_id)
        self.assertIsInstance(resp["tags"], list)
        resp_tags_id = [t["id"] for t in resp["tags"]]
        self.assertEquals(1, len(resp_tags_id))
        self.assertFalse(tag_id in resp_tags_id)
        await self._delete_entry(entry_id)

    async def _create_entry(self):
        data = self.entry_data.copy()
        data["title"] = "Some title 2"
        data["url"] = f"https://somwhere.over.the.raibow.com/{uuid.uuid4()}"
        entry = await self.w.post_entries(**data)
        self.assertIsInstance(entry["id"], int)
        return entry["id"]

    async def _delete_entry(self, entry_id):
        await self.w.delete_entries(entry_id)


if __name__ == "__main__":
    unittest.main()