Fixes API unit tests
Fix all unit tests, and use a `testcfg.ini` file to load test configuration values (e.g. api_client, url, ...)
This commit is contained in:
@@ -1,121 +1,181 @@
|
||||
# coding: utf-8
|
||||
import asyncio
|
||||
import configparser
|
||||
import datetime
|
||||
import os
|
||||
import sys
|
||||
import uuid
|
||||
import unittest
|
||||
from wallabag import Wallabag
|
||||
|
||||
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):
|
||||
|
||||
host = 'http://localhost:8000'
|
||||
client_id = ''
|
||||
client_secret = ''
|
||||
token = ''
|
||||
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",
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
access_token = self.test_get_token()
|
||||
self.format = 'json'
|
||||
self.w = Wallabag(host=self.host,
|
||||
token=access_token,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret)
|
||||
@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)
|
||||
|
||||
def test_get_token(self):
|
||||
params = {"grant_type": "password",
|
||||
"client_id":
|
||||
'1_4wqe1riwt0qoks844kwc4go08koogkgk88go4cckkwg0408kcg',
|
||||
"client_secret": '4mzw3qwi1xyc0cks4k80s4c8kco40wwkkkw0g40kwk4o4c44co',
|
||||
"username": 'wallabag',
|
||||
"password": 'wallabag'}
|
||||
print(self.host)
|
||||
data = Wallabag.get_token(host=self.host, **params)
|
||||
print(data)
|
||||
self.assertTrue(isinstance(data, str), True)
|
||||
return 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())
|
||||
|
||||
def create_entry(self):
|
||||
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'
|
||||
data = self.w.post_entries(url, title, tags, starred, archive, content, language, published_at, authors,
|
||||
public, original_url)
|
||||
@run_async
|
||||
async def test_get_token(self):
|
||||
data = await Wallabag.get_token(**self._configuration)
|
||||
print(data, file=sys.stderr)
|
||||
self.assertIsInstance(data, str)
|
||||
|
||||
return data
|
||||
|
||||
def test_get_entries(self):
|
||||
params = {'delete': 0,
|
||||
'sort': 'created',
|
||||
'order': 'desc',
|
||||
'page': 1,
|
||||
'perPage': 30,
|
||||
'tags': []}
|
||||
data = self.w.get_entries(**params)
|
||||
@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)
|
||||
|
||||
def test_get_entry(self):
|
||||
entry = 1
|
||||
self.assertTrue(isinstance(entry, int), True)
|
||||
data = self.w.get_entry(entry)
|
||||
self.assertTrue(data, str)
|
||||
@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)
|
||||
|
||||
def test_get_entry_tags(self):
|
||||
entry = 1
|
||||
self.assertTrue(isinstance(entry, int), True)
|
||||
data = self.w.get_entry_tags(entry)
|
||||
@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)
|
||||
|
||||
def test_get_tags(self):
|
||||
data = 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"])
|
||||
|
||||
def test_post_entries(self):
|
||||
data = self.create_entry()
|
||||
@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)
|
||||
|
||||
def test_patch_entries(self):
|
||||
entry = 1
|
||||
params = {'title': 'I change the title',
|
||||
'archive': 0,
|
||||
'tags': ["bimbo", "pipo"],
|
||||
'order': 'asc',
|
||||
'star': 0,
|
||||
'delete': 0}
|
||||
self.assertTrue(isinstance(entry, int), True)
|
||||
self.assertTrue(isinstance(params, dict), True)
|
||||
data = self.w.patch_entries(entry, **params)
|
||||
@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)
|
||||
|
||||
def test_delete_entries(self):
|
||||
entry = self.create_entry()
|
||||
self.assertTrue(isinstance(entry['id'], int), True)
|
||||
data = self.w.delete_entries(entry['id'])
|
||||
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)
|
||||
|
||||
def test_post_entry_tags(self):
|
||||
entry = 1
|
||||
self.assertTrue(isinstance(entry, int), True)
|
||||
tags = ['foo', 'bar']
|
||||
self.assertTrue(isinstance(tags, list), True)
|
||||
data = self.w.post_entry_tags(entry, tags)
|
||||
self.assertTrue(data, True)
|
||||
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"]
|
||||
|
||||
"""
|
||||
def test_delete_entry_tag(self):
|
||||
entry = self.create_entry()
|
||||
tag = 'bar'
|
||||
self.assertTrue(isinstance(entry['id'], int), True)
|
||||
self.assertTrue(isinstance(tag, str), True)
|
||||
resp = self.w.delete_entry_tag(entry['id'], tag)
|
||||
self.assertTrue(resp, True)
|
||||
"""
|
||||
async def _delete_entry(self, entry_id):
|
||||
await self.w.delete_entries(entry_id)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user