-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/tests/data/dialogs/auth_dialog.html b/tests/data/dialogs/auth_dialog.html
deleted file mode 100644
index b4b5d92..0000000
--- a/tests/data/dialogs/auth_dialog.html
+++ /dev/null
@@ -1,218 +0,0 @@
-
-
-
-
Mail.Ru — Запрос доступа
-
-
-
-
-
-
-
-
-
-
Необходим доступ к вашим данным
-
-
App name просит вас разрешить следующие действия:
-
- - Получать информацию о вашей анкете, списке друзей, фотографиях и музыке
- - Создавать альбомы и размещать там изображения
- - Отправлять личные сообщения другим пользователям
-
-
-
-
-
-
-
-
-
\ No newline at end of file
diff --git a/tests/test_scrapers.py b/tests/test_scrapers.py
deleted file mode 100644
index 5799085..0000000
--- a/tests/test_scrapers.py
+++ /dev/null
@@ -1,97 +0,0 @@
-import atexit
-import os
-
-import pytest
-
-from aiomailru.sessions import ImplicitSession, TokenSession
-
-
-EMAIL = os.environ.get('MAILRU_EMAIL')
-PASSWD = os.environ.get('MAILRU_PASSWD')
-CLIENT_ID = os.environ.get('MAILRU_CLIENT_ID')
-PRIVATE_KEY = os.environ.get('MAILRU_PRIVATE_KEY')
-SECRET_KEY = os.environ.get('MAILRU_SECRET_KEY')
-SCOPE = os.environ.get('MAILRU_SCOPE')
-
-BROWSER_ENDPOINT = os.environ.get('PYPPETEER_BROWSER_ENDPOINT')
-
-skip_scrapers = False
-try:
- from aiomailru.scrapers import APIScraper
-except ModuleNotFoundError:
- reasons = ['pyppeteer not found']
-else:
- reasons = []
- if EMAIL is None:
- reasons.append('MAILRU_EMAIL (user e-mail) not set')
- if PASSWD is None:
- reasons.append('MAILRU_PASSWD (password) not set')
- if CLIENT_ID is None:
- reasons.append('MAILRU_CLIENT_ID (app id) not set')
- if PRIVATE_KEY is None:
- reasons.append('MAILRU_PRIVATE_KEY not set')
- if SECRET_KEY is None:
- reasons.append('MAILRU_SECRET_KEY not set')
- if SCOPE is None:
- reasons.append('MAILRU_SCOPE (permissions) not set')
- if BROWSER_ENDPOINT is None:
- reasons.append('PYPPETEER_BROWSER_ENDPOINT (Chrome endpoint) not set')
-
-
-if reasons:
- skip_scrapers = True
- atexit.register(lambda: print('\n'.join(reasons)))
- atexit.register(lambda: print('Scrapers tests were skipped'))
-
-
-GROUP_ID = os.environ.get(
- 'MAILRU_GROUP_ID',
- '5396991818946538245' # My@Mail.Ru official community
-)
-
-
-@pytest.mark.skipif(skip_scrapers, reason=';'.join(reasons))
-class TestScrapers:
-
- @pytest.fixture
- def app(self):
- return CLIENT_ID, PRIVATE_KEY, SECRET_KEY
-
- @pytest.fixture
- def cred(self):
- return EMAIL, PASSWD, SCOPE
-
- @pytest.fixture
- async def session(self, app, cred):
- async with ImplicitSession(*app, *cred) as session:
- token = session.session_key
- cookies = session.cookies
- return TokenSession(*app, token, 0, cookies=cookies)
-
- @pytest.mark.asyncio
- async def test_groups_get(self, session: TokenSession):
- async with session:
- api = APIScraper(session)
- _ = await api.groups.get(scrape=True)
- await api.browser.disconnect()
-
- @pytest.mark.asyncio
- async def test_groups_get_info(self, session: TokenSession):
- async with session:
- api = APIScraper(session)
- _ = await api.groups.getInfo(uids=GROUP_ID, scrape=True)
- await api.browser.disconnect()
-
- @pytest.mark.asyncio
- async def test_groups_join(self, session: TokenSession):
- async with session:
- api = APIScraper(session)
- _ = await api.groups.join(group_id=GROUP_ID, scrape=True)
- await api.browser.disconnect()
-
- @pytest.mark.asyncio
- async def test_stream_get_by_author(self, session: TokenSession):
- async with session:
- api = APIScraper(session)
- _ = await api.stream.getByAuthor(uid=GROUP_ID, scrape=True)
- await api.browser.disconnect()
diff --git a/tests/test_sessions.py b/tests/test_sessions.py
index 6cf814c..305aa45 100644
--- a/tests/test_sessions.py
+++ b/tests/test_sessions.py
@@ -1,70 +1,27 @@
-import json
-from urllib.parse import unquote
-
+"""Sessions tests."""
import pytest
+from httpx import HTTPStatusError
-from aiomailru.exceptions import (
- Error, OAuthError, APIError, EmptyResponseError
-)
-from aiomailru.utils import SignatureCircuit
-from aiomailru.sessions import (
- PublicSession,
- TokenSession,
- ImplicitSession,
-)
+from aiomailru.session import PublicSession, TokenSession
class TestPublicSession:
- """Tests of PublicSession class."""
-
- @pytest.mark.asyncio
- async def test_error_request(self, error_server, error):
- async with PublicSession() as session:
- session.PUBLIC_URL = error_server.url
-
- session.pass_error = True
- response = await session.public_request()
- assert response == error
-
- @pytest.mark.asyncio
- async def test_error_request_with_raising(self, error_server):
- async with PublicSession() as session:
- session.PUBLIC_URL = error_server.url
-
- session.pass_error = False
- with pytest.raises(APIError):
- await session.public_request()
-
- @pytest.mark.asyncio
- async def test_dummy_request(self, dummy_server, dummy):
- async with PublicSession() as session:
- session.PUBLIC_URL = dummy_server.url
+ """Tests for PublicSession class."""
- session.pass_error = True
- response = await session.public_request()
- assert response == dummy
+ async def test_failed_request(self, error_server):
+ """Test failed request."""
+ session = PublicSession()
+ session.client.base_url = error_server.url
- @pytest.mark.asyncio
- async def test_dummy_request_with_raising(self, dummy_server):
- async with PublicSession() as session:
- session.PUBLIC_URL = dummy_server.url
+ with pytest.raises(HTTPStatusError):
+ await session.request({})
- session.pass_error = False
- with pytest.raises(EmptyResponseError):
- await session.public_request()
+ async def test_regulat_request(self, data_server):
+ """Test regular request."""
+ session = PublicSession()
+ session.client.base_url = data_server.url
- @pytest.mark.asyncio
- async def test_data_request(self, data_server, data):
- async with PublicSession() as session:
- session.PUBLIC_URL = data_server.url
-
- session.pass_error = True
- response = await session.public_request()
- assert response == data
-
- session.pass_error = False
- response = await session.public_request()
- assert response == data
+ assert await session.request({}) == {'key': 'value'}
class TestTokenSession:
@@ -72,201 +29,37 @@ class TestTokenSession:
@pytest.fixture
def app(self):
+ """Return app info."""
return {'app_id': 123, 'private_key': '', 'secret_key': ''}
@pytest.fixture
def token(self):
+ """Return token info."""
return {'access_token': '', 'uid': 0}
- @pytest.mark.asyncio
- async def test_sig_circuit(self, app, token):
- async with TokenSession(**app, **token) as session:
- assert session.sig_circuit is SignatureCircuit.UNDEFINED
-
- session.secret_key = 'secret key'
- assert session.sig_circuit is SignatureCircuit.SERVER_SERVER
-
- session.uid = 456
- session.private_key = 'private key'
- assert session.sig_circuit is SignatureCircuit.CLIENT_SERVER
-
- @pytest.mark.asyncio
- async def test_required_params(self, app, token):
- async with TokenSession(**app, **token) as session:
- assert 'app_id' in session.required_params
- assert 'session_key' in session.required_params
- assert 'secure' not in session.required_params
- session.uid = 456
- session.private_key = ''
- session.secret_key = 'secret key'
- assert 'secure' in session.required_params
-
- @pytest.mark.asyncio
- async def test_params_to_str(self, app, token):
- async with TokenSession(**app, **token) as session:
- params = {'"a"': 1, '"b"': 2, '"c"': 3}
-
- session.uid = 789
- session.private_key = 'private key'
- query = session.params_to_str(params)
- assert query == '789"a"=1"b"=2"c"=3private key'
-
- session.uid = 0
- session.private_key = ''
- session.secret_key = 'secret key'
- query = session.params_to_str(params)
- assert query == '"a"=1"b"=2"c"=3secret key'
-
- session.secret_key = ''
- with pytest.raises(Error):
- _ = session.params_to_str(params)
-
- @pytest.mark.asyncio
- async def test_error_request(self, app, token, error_server, error):
- async with TokenSession(**app, **token) as session:
- session.API_URL = error_server.url
- session.secret_key = 'secret key'
-
- session.pass_error = True
- response = await session.request(params={'key': 'value'})
- assert response == error
-
- @pytest.mark.asyncio
- async def test_error_request_with_raising(self, app, token, error_server):
- async with TokenSession(**app, **token) as session:
- session.API_URL = error_server.url
- session.secret_key = 'secret key'
-
- session.pass_error = False
- with pytest.raises(APIError):
- await session.request(params={'key': 'value'})
-
- @pytest.mark.asyncio
- async def test_dummy_request(self, app, token, dummy_server, dummy):
- async with TokenSession(**app, **token) as session:
- session.API_URL = dummy_server.url
- session.secret_key = 'secret key'
-
- session.pass_error = True
- response = await session.request(params={'key': 'value'})
- assert response == dummy
-
- @pytest.mark.asyncio
- async def test_dummy_request_with_raising(self, app, token, dummy_server):
- async with TokenSession(**app, **token) as session:
- session.API_URL = dummy_server.url
- session.secret_key = 'secret key'
-
- session.pass_error = False
- with pytest.raises(EmptyResponseError):
- await session.request(params={'key': 'value'})
-
- @pytest.mark.asyncio
- async def test_data_request(self, app, token, data_server, data):
- async with TokenSession(**app, **token) as session:
- session.API_URL = data_server.url
- session.secret_key = 'secret key'
-
- session.pass_error = True
- response = await session.request(params={'key': 'value'})
- assert response == data
-
- session.pass_error = False
- response = await session.request(params={'key': 'value'})
- assert response == data
-
-
-class TestImplicitSession:
- """Tests of ImplicitSession class."""
-
- @pytest.fixture
- def app(self):
- return {'app_id': 123, 'private_key': '', 'secret_key': ''}
-
- @pytest.fixture
- def cred(self):
- return {
- 'email': 'email@example.ru',
- 'passwd': 'password',
- 'scope': 'permission1 permission2 permission3',
- }
-
- @pytest.mark.asyncio
- async def test_get_auth_dialog(self, app, cred, httpserver, auth_dialog):
- # success
- httpserver.serve_content(**{
- 'code': 200,
- 'headers': {'Content-Type': 'text/html'},
- 'content': auth_dialog,
- })
- session = ImplicitSession(**app, **cred)
- session.OAUTH_URL = httpserver.url
- url, html = await session._get_auth_dialog()
-
- assert url.query['client_id'] == str(session.app_id)
- assert url.query['redirect_uri'] == unquote(session.REDIRECT_URI)
- assert url.query['response_type'] == session.params['response_type']
- assert url.query['scope'] == session.scope
- assert html == auth_dialog
-
- # fail
- httpserver.serve_content(**{
- 'code': 400,
- 'headers': {'Content-Type': 'text/json'},
- 'content': json.dumps({'error': '', 'error_description': ''})
- })
- with pytest.raises(OAuthError):
- _ = await session._get_auth_dialog()
-
- await session.close()
-
- @pytest.mark.asyncio
- async def test_post_auth_dialog(self, app, cred, httpserver,
- auth_dialog, access_dialog):
- # success
- httpserver.serve_content(**{'code': 200, 'content': access_dialog})
- session = ImplicitSession(**app, **cred)
-
- auth_dialog = auth_dialog.replace(
- 'https://auth.mail.ru/cgi-bin/auth', httpserver.url,
+ async def test_error_request_with_raising(self, error_server):
+ """Test error request that raises an error."""
+ session = TokenSession(
+ app_id='123',
+ session_key='session key',
+ secret='secret key',
+ secure='1',
+ uid='',
)
- url, html = await session._post_auth_dialog(auth_dialog)
- assert html == access_dialog
-
- # fail
- httpserver.serve_content(**{'code': 400, 'content': ''})
- with pytest.raises(OAuthError):
- _ = await session._post_auth_dialog(auth_dialog)
-
- await session.close()
-
- @pytest.mark.asyncio
- async def test_post_access_dialog(self, app, cred, httpserver, access_dialog):
- # success
- httpserver.serve_content(**{'code': 200, 'content': 'blank page'})
- session = ImplicitSession(**app, **cred)
-
- access_dialog = access_dialog.replace(
- 'https://connect.mail.ru/oauth/authorize', httpserver.url
+ session.client.base_url = error_server.url
+
+ with pytest.raises(HTTPStatusError):
+ await session.request(params={'key': 'value'})
+
+ async def test_data_request(self, data_server):
+ """Test regular request."""
+ session = TokenSession(
+ app_id='123',
+ session_key='session key',
+ secret='secret key',
+ secure='1',
+ uid='',
)
- url, html = await session._post_access_dialog(access_dialog)
- assert html == 'blank page'
-
- # fail
- httpserver.serve_content(**{'code': 400, 'content': ''})
- with pytest.raises(OAuthError):
- _ = await session._post_access_dialog(access_dialog)
-
- await session.close()
-
- @pytest.mark.asyncio
- async def test_get_access_token(self, app, cred, httpserver):
- # fail
- httpserver.serve_content(**{'code': 400, 'content': ''})
- session = ImplicitSession(**app, **cred)
- session.OAUTH_URL = httpserver.url
-
- with pytest.raises(OAuthError):
- _ = await session._get_access_token()
+ session.client.base_url = data_server.url
- await session.close()
+ assert await session.request(params={'k': 'v'}) == {'key': 'value'}