mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-07 01:41:17 +00:00
Add Logger Tests (#1707)
* Added openbb_terminal/logger tests * Added tests for path_tracking * Just in case lol * Finished test_directories * Added more tests * Fixed flake8 error * Fixed flake8 error * Coverage for test_expired_files * Added handling for no boto3 * Added test mock * Added test mock * Added fix * Test fix * Fix for windows * Added int * More tests * Added fixes * Fixed black issues * Tests : update fixture examples * Tests : core/log/generation * Core : log/generation * replaced conftest delete with temp_path * Fixed mocks * refactored get settings to use tmp_path * reverted codespell * Making random faillign test linux only * Swapped to fixture * Fixed conftest * Logs collection : add docstring + clean lines * Tests : fix windows specific issue * Tests : skip unsound tests which changes logger state * Black * Linting Co-authored-by: Chavithra PARANA <chavithra@gmail.com> Co-authored-by: didierlopes.eth <dro.lopes@campus.fct.unl.pt>
This commit is contained in:
@@ -73,8 +73,7 @@ def send_to_s3_using_presigned_url(
|
||||
|
||||
raw_response.raise_for_status()
|
||||
|
||||
if raw_response.status_code == 204:
|
||||
pass
|
||||
# SUCCESS IF STATUS_CODE == 204
|
||||
|
||||
|
||||
def send_to_s3(
|
||||
@@ -85,6 +84,28 @@ def send_to_s3(
|
||||
tmp_file: Path,
|
||||
last: bool = False,
|
||||
):
|
||||
"""Send a file into a s3 bucket.
|
||||
|
||||
Args:
|
||||
archives_file (Path):
|
||||
Destination Path after processing.
|
||||
aws_settings (AWSSettings):
|
||||
AWS settings.
|
||||
file (Path):
|
||||
Path of the file to process.
|
||||
object_key (str): _description_
|
||||
File location inside the s3 bucket.
|
||||
tmp_file (Path):
|
||||
Temporary Path in which to put the file during processing.
|
||||
last (bool, optional):
|
||||
Whether or not this is the last sending before program exit.
|
||||
Defaults to False.
|
||||
|
||||
Raises:
|
||||
AttributeError:
|
||||
If `file` is empty.
|
||||
"""
|
||||
|
||||
api_url = DEFAULT_API_URL
|
||||
bucket = DEFAULT_BUCKET
|
||||
|
||||
|
||||
@@ -9,12 +9,12 @@ from typing import List
|
||||
# IMPORTATION INTERNAL
|
||||
|
||||
|
||||
def get_timestamp_from_x_days(x: int):
|
||||
def get_timestamp_from_x_days(x: int) -> float:
|
||||
timestamp_from_x_days = datetime.now().timestamp() - x * 86400
|
||||
return timestamp_from_x_days
|
||||
|
||||
|
||||
def get_expired_file_list(directory: Path, before_timestamp: float):
|
||||
def get_expired_file_list(directory: Path, before_timestamp: float) -> List[Path]:
|
||||
expired_files = list()
|
||||
if directory.exists and directory.is_dir():
|
||||
for file in directory.iterdir():
|
||||
|
||||
3
tests/FIXTURES.md
vendored
3
tests/FIXTURES.md
vendored
@@ -254,6 +254,7 @@ You can't combine these two fixtures :
|
||||
|
||||
Example of usage :
|
||||
```python
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
@pytest.mark.record_stdout
|
||||
@@ -262,11 +263,13 @@ def test_function(recorder):
|
||||
some_list = [7, 8, 9 ]
|
||||
some_tuple = (4, 5, 6)
|
||||
some_string = "Some string"
|
||||
some_complex_tuple = (pd.DataFrame(), some_dict)
|
||||
|
||||
recorder.capture(some_dict)
|
||||
recorder.capture(some_list)
|
||||
recorder.capture(some_string)
|
||||
recorder.capture(some_tuple)
|
||||
recorder.capture_list(some_complex_tuple)
|
||||
```
|
||||
|
||||
This will generate one or multiple text file(s) to store the `captured` variables.
|
||||
|
||||
45
tests/openbb_terminal/core/log/collection/cassettes/test_s3_sender/test_fetch_presigned_url.yaml
vendored
Normal file
45
tests/openbb_terminal/core/log/collection/cassettes/test_s3_sender/test_fetch_presigned_url.yaml
vendored
Normal file
@@ -0,0 +1,45 @@
|
||||
interactions:
|
||||
- request:
|
||||
body: '{"object_key": "gst-app/logs/MOCK_OBJECT_KEY"}'
|
||||
headers:
|
||||
Accept:
|
||||
- '*/*'
|
||||
Accept-Encoding:
|
||||
- gzip, deflate
|
||||
Connection:
|
||||
- keep-alive
|
||||
Content-Length:
|
||||
- '46'
|
||||
Content-Type:
|
||||
- application/json
|
||||
User-Agent:
|
||||
- python-requests/2.27.1
|
||||
method: PUT
|
||||
uri: https://gst.openbb.app/logging
|
||||
response:
|
||||
body:
|
||||
string: '{"url": "https://gst-logging.s3.amazonaws.com/", "fields": {"key":
|
||||
"gst-app/logs/MOCK_OBJECT_KEY", "x-amz-algorithm": "AWS4-HMAC-SHA256", "x-amz-credential":
|
||||
"ASIAWJBK4EZQMTEBW2WN/20220503/us-east-2/s3/aws4_request", "x-amz-date": "20220503T215729Z",
|
||||
"x-amz-security-token": "IQoJb3JpZ2luX2VjEJ7//////////wEaCXVzLWVhc3QtMiJGMEQCIGmU7C0zpaILkcHWPqd6vRMs7xy3rv489xd+uL7HMRhdAiBe+OIdNrOUxFxIxVQtkUGphO5wTX7pJ6lI2dIF2CjfTCqcAghnEAAaDDQzMTczNDEzODQ2NCIMMqpXsS4vwbSlg8EGKvkBd06m3Bwq072qUtVLztwvi4aNweRy59C9IpCKYOOp2TQFl29EBuvLG467HdXlIzTOMJhIu12xx8ngpMl5d7OR4zevsOPcGv8LSm3/6o4DnJN8v2cUlAMKNv6Bk2M11e+jbJHtb6wJDhZfz2nR6oQaZnh2uYoB4im9nObJDTvkArx/jdocA9Qi2n1x4RiH6J4AC6n+0S2jcAX7M71gj1rRLXT3IPboA2XU+XAetfKhri2moXntqXe2avFM0WRM8SFkmf1d5g23F7cSsy623atqKx8CUV4H+D4nnb4dz0EH1TnmPMBS4583SPMoFdzIIVdaLBg6K1GTUjmdMMnKxpMGOpsBTZzmyUy+Idvv7NeC29GXwiZizXVwb+CmQk1IC6oNX6ghwF5BihS/10OWfH2dpRrRJCimyy05bnhLHYlKsulDHnJ8UVad/ohjaoz6hpDk6XI8iexfF8LyXn7qmGl6ncuw8wIsLDBlD2otd58vWoEld8g5REkYlSWJ8jN8ejomYZMwEW4iD9EI7D+0GRARG7PowV2/EcuiFl+BzGA=",
|
||||
"policy": "eyJleHBpcmF0aW9uIjogIjIwMjItMDUtMDNUMjI6MDI6MjlaIiwgImNvbmRpdGlvbnMiOiBbeyJidWNrZXQiOiAiZ3N0LWxvZ2dpbmcifSwgeyJrZXkiOiAiZ3N0LWFwcC9sb2dzL01PQ0tfT0JKRUNUX0tFWSJ9LCB7IngtYW16LWFsZ29yaXRobSI6ICJBV1M0LUhNQUMtU0hBMjU2In0sIHsieC1hbXotY3JlZGVudGlhbCI6ICJBU0lBV0pCSzRFWlFNVEVCVzJXTi8yMDIyMDUwMy91cy1lYXN0LTIvczMvYXdzNF9yZXF1ZXN0In0sIHsieC1hbXotZGF0ZSI6ICIyMDIyMDUwM1QyMTU3MjlaIn0sIHsieC1hbXotc2VjdXJpdHktdG9rZW4iOiAiSVFvSmIzSnBaMmx1WDJWakVKNy8vLy8vLy8vLy93RWFDWFZ6TFdWaGMzUXRNaUpHTUVRQ0lHbVU3QzB6cGFJTGtjSFdQcWQ2dlJNczd4eTNydjQ4OXhkK3VMN0hNUmhkQWlCZStPSWROck9VeEZ4SXhWUXRrVUdwaE81d1RYN3BKNmxJMmRJRjJDamZUQ3FjQWdobkVBQWFERFF6TVRjek5ERXpPRFEyTkNJTU1xcFhzUzR2d2JTbGc4RUdLdmtCZDA2bTNCd3EwNzJxVXRWTHp0d3ZpNGFOd2VSeTU5QzlJcENLWU9PcDJUUUZsMjlFQnV2TEc0NjdIZFhsSXpUT01KaEl1MTJ4eDhuZ3BNbDVkN09SNHpldnNPUGNHdjhMU20zLzZvNERuSk44djJjVWxBTUtOdjZCazJNMTFlK2piSkh0YjZ3SkRoWmZ6Mm5SNm9RYVpuaDJ1WW9CNGltOW5PYkpEVHZrQXJ4L2pkb2NBOVFpMm4xeDRSaUg2SjRBQzZuKzBTMmpjQVg3TTcxZ2oxclJMWFQzSVBib0EyWFUrWEFldGZLaHJpMm1vWG50cVhlMmF2Rk0wV1JNOFNGa21mMWQ1ZzIzRjdjU3N5NjIzYXRxS3g4Q1VWNEgrRDRubmI0ZHowRUgxVG5tUE1CUzQ1ODNTUE1vRmR6SUlWZGFMQmc2SzFHVFVqbWRNTW5LeHBNR09wc0JUWnpteVV5K0lkdnY3TmVDMjlHWHdpWml6WFZ3YitDbVFrMUlDNm9OWDZnaHdGNUJpaFMvMTBPV2ZIMmRwUnJSSkNpbXl5MDVibmhMSFlsS3N1bERIbko4VVZhZC9vaGphb3o2aHBEazZYSThpZXhmRjhMeVhuN3FtR2w2bmN1dzh3SXNMREJsRDJvdGQ1OHZXb0VsZDhnNVJFa1lsU1dKOGpOOGVqb21ZWk13RVc0aUQ5RUk3RCswR1JBUkc3UG93VjIvRWN1aUZsK0J6R0E9In1dfQ==",
|
||||
"x-amz-signature": "8ce6f27a0ca0e673cc12cbe512c3e21cb4156e2693a8b5194acb1d855cfa524a"}}'
|
||||
headers:
|
||||
Connection:
|
||||
- keep-alive
|
||||
Content-Length:
|
||||
- '2532'
|
||||
Content-Type:
|
||||
- application/json
|
||||
Date:
|
||||
- Tue, 03 May 2022 21:57:29 GMT
|
||||
X-Amzn-Trace-Id:
|
||||
- Root=1-6271a549-1ec446d37f7c3f607c1153a4;Sampled=0
|
||||
x-amz-apigw-id:
|
||||
- RkbDeEMgCYcFu0Q=
|
||||
x-amzn-RequestId:
|
||||
- 02208106-c09e-4d62-82cf-a3ee313b084b
|
||||
status:
|
||||
code: 200
|
||||
message: OK
|
||||
version: 1
|
||||
20
tests/openbb_terminal/core/log/collection/test_log_sender.py
Normal file
20
tests/openbb_terminal/core/log/collection/test_log_sender.py
Normal file
@@ -0,0 +1,20 @@
|
||||
from openbb_terminal.core.log.collection import log_sender
|
||||
|
||||
|
||||
def test_queue_str(tmp_path):
|
||||
log_sender.QueueItem(tmp_path).__str__()
|
||||
|
||||
|
||||
def test_sender_settings(settings):
|
||||
value = log_sender.LogSender(settings).settings
|
||||
assert value is not None
|
||||
|
||||
|
||||
def test_sender_fails(settings):
|
||||
value = log_sender.LogSender(settings).fails
|
||||
assert value is not None
|
||||
|
||||
|
||||
def test_sender_queue(settings):
|
||||
value = log_sender.LogSender(settings).queue
|
||||
assert value is not None
|
||||
@@ -0,0 +1,44 @@
|
||||
from datetime import datetime
|
||||
import pytest
|
||||
from openbb_terminal.core.log.collection import logging_clock
|
||||
|
||||
|
||||
clock = logging_clock.LoggingClock()
|
||||
now = datetime.now()
|
||||
|
||||
|
||||
def mock_next(**_):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"precision", [logging_clock.Precision.hour, logging_clock.Precision.minute]
|
||||
)
|
||||
def test_calculate_next_sharp(precision):
|
||||
value = clock.calculate_next_sharp(now, precision)
|
||||
assert value
|
||||
|
||||
|
||||
def test_calculate_next_sharp_invalid():
|
||||
with pytest.raises(Exception):
|
||||
clock.calculate_next_sharp(now, "bad")
|
||||
|
||||
|
||||
# TODO: find a better way to mock the while loop
|
||||
def test_do_action_every_sharp(mocker):
|
||||
mock = mocker.Mock()
|
||||
mock.count = 0
|
||||
mock.mock_next = mock_next
|
||||
with pytest.raises(NotImplementedError):
|
||||
clock.do_action_every_sharp(mock.mock_next)
|
||||
|
||||
|
||||
def test_run(mocker):
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.collection.logging_clock.LoggingClock.do_action_every_sharp"
|
||||
)
|
||||
clock.run()
|
||||
|
||||
|
||||
def test_default_action():
|
||||
clock.default_action()
|
||||
168
tests/openbb_terminal/core/log/collection/test_s3_sender.py
Normal file
168
tests/openbb_terminal/core/log/collection/test_s3_sender.py
Normal file
@@ -0,0 +1,168 @@
|
||||
# IMPORTATION STANDARD
|
||||
from pathlib import Path
|
||||
|
||||
# IMPORTATION THIRDPARTY
|
||||
import pytest
|
||||
|
||||
# IMPORTATION INTERNAL
|
||||
from openbb_terminal.core.log.generation.settings import AWSSettings
|
||||
from openbb_terminal.core.log.constants import DEFAULT_API_URL
|
||||
from openbb_terminal.core.log.collection import s3_sender
|
||||
|
||||
# pylint: disable=W0611
|
||||
|
||||
|
||||
@pytest.mark.vcr(record_mode="none")
|
||||
def test_send_to_s3_directly(mocker):
|
||||
s3_sender.boto3 = mocker.Mock()
|
||||
mocker.patch.object(
|
||||
target=s3_sender,
|
||||
attribute="WITH_BOTO3",
|
||||
new=True,
|
||||
)
|
||||
|
||||
aws_access_key_id = "MOCK_ACCESS_KEY_ID"
|
||||
aws_secret_access_key = "MOCK_ACCESS_SECRET_KEY"
|
||||
bucket = "MOCK_BUCKET"
|
||||
file = Path(__file__)
|
||||
object_key = "MOCK_S3/OBJECT_KEY"
|
||||
|
||||
s3_sender.send_to_s3_directly(
|
||||
aws_access_key_id=aws_access_key_id,
|
||||
aws_secret_access_key=aws_secret_access_key,
|
||||
bucket=bucket,
|
||||
file=file,
|
||||
object_key=object_key,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.vcr(record_mode="none")
|
||||
def test_send_to_s3_directly_exception(mocker):
|
||||
s3_sender.boto3 = mocker.Mock()
|
||||
mocker.patch.object(
|
||||
target=s3_sender,
|
||||
attribute="WITH_BOTO3",
|
||||
new=False,
|
||||
)
|
||||
|
||||
aws_access_key_id = "MOCK_ACCESS_KEY_ID"
|
||||
aws_secret_access_key = "MOCK_ACCESS_SECRET_KEY"
|
||||
bucket = "MOCK_BUCKET"
|
||||
file = Path(__file__)
|
||||
object_key = "MOCK_S3/OBJECT_KEY"
|
||||
|
||||
with pytest.raises(ModuleNotFoundError):
|
||||
s3_sender.send_to_s3_directly(
|
||||
aws_access_key_id=aws_access_key_id,
|
||||
aws_secret_access_key=aws_secret_access_key,
|
||||
bucket=bucket,
|
||||
file=file,
|
||||
object_key=object_key,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.vcr
|
||||
def test_fetch_presigned_url():
|
||||
object_key = "gst-app/logs/MOCK_OBJECT_KEY"
|
||||
s3_sender.fetch_presigned_url(api_url=DEFAULT_API_URL, object_key=object_key)
|
||||
|
||||
|
||||
@pytest.mark.vcr(record_mode="none")
|
||||
def test_send_to_s3_using_presigned_url(mocker, tmp_path):
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.collection.s3_sender.fetch_presigned_url",
|
||||
return_value={"fields": [1, 2, 3], "url": "http://"},
|
||||
)
|
||||
mocker.patch("openbb_terminal.core.log.collection.s3_sender.requests")
|
||||
|
||||
api_url = DEFAULT_API_URL
|
||||
file = tmp_path.joinpath("mock_log_file")
|
||||
object_key = "MOCK_S3/OBJECT_KEY"
|
||||
|
||||
with open(file, "w") as f:
|
||||
f.write("Mocking a log file to send to s3.")
|
||||
|
||||
s3_sender.send_to_s3_using_presigned_url(
|
||||
api_url=api_url,
|
||||
file=file,
|
||||
object_key=object_key,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.vcr(record_mode="none")
|
||||
@pytest.mark.parametrize(
|
||||
"aws_settings, last",
|
||||
[
|
||||
(
|
||||
AWSSettings(
|
||||
aws_access_key_id="MOCK_AWS_ACCESS_KEY_ID",
|
||||
aws_secret_access_key="MOCK_AWS_ACCESS_KEY",
|
||||
),
|
||||
False,
|
||||
),
|
||||
(
|
||||
AWSSettings(
|
||||
aws_access_key_id="REPLACE_ME",
|
||||
aws_secret_access_key="REPLACE_ME",
|
||||
),
|
||||
True,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_send_to_s3(aws_settings, mocker, last, tmp_path):
|
||||
mocker.patch("openbb_terminal.core.log.collection.s3_sender.send_to_s3_directly")
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.collection.s3_sender.send_to_s3_using_presigned_url"
|
||||
)
|
||||
|
||||
archives_file = tmp_path.joinpath("archives").joinpath("mock_log_file")
|
||||
tmp_file = tmp_path.joinpath("tmp").joinpath("mock_log_file")
|
||||
file = tmp_path.joinpath("mock_log_file")
|
||||
object_key = "MOCK_S3/OBJECT_KEY"
|
||||
|
||||
with open(file, "w") as f:
|
||||
f.write("Mocking a log file to send to s3.")
|
||||
|
||||
assert file.exists()
|
||||
assert not archives_file.exists()
|
||||
assert not tmp_file.exists()
|
||||
|
||||
s3_sender.send_to_s3(
|
||||
archives_file=archives_file,
|
||||
aws_settings=aws_settings,
|
||||
file=file,
|
||||
object_key=object_key,
|
||||
tmp_file=tmp_file,
|
||||
last=last,
|
||||
)
|
||||
|
||||
assert last or not file.exists()
|
||||
assert archives_file.exists()
|
||||
assert not tmp_file.exists()
|
||||
|
||||
|
||||
@pytest.mark.vcr(record_mode="none")
|
||||
def test_send_to_s3_exception(mocker, tmp_path):
|
||||
mocker.patch("openbb_terminal.core.log.collection.s3_sender.send_to_s3_directly")
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.collection.s3_sender.send_to_s3_using_presigned_url"
|
||||
)
|
||||
|
||||
archives_file = tmp_path.joinpath("archives").joinpath("mock_log_file")
|
||||
tmp_file = tmp_path.joinpath("tmp").joinpath("mock_log_file")
|
||||
file = tmp_path.joinpath("mock_log_file")
|
||||
file.touch()
|
||||
object_key = "MOCK_S3/OBJECT_KEY"
|
||||
|
||||
with pytest.raises(AttributeError):
|
||||
s3_sender.send_to_s3(
|
||||
archives_file=archives_file,
|
||||
aws_settings=AWSSettings(
|
||||
aws_access_key_id="REPLACE_ME",
|
||||
aws_secret_access_key="REPLACE_ME",
|
||||
),
|
||||
file=file,
|
||||
object_key=object_key,
|
||||
tmp_file=tmp_file,
|
||||
last=False,
|
||||
)
|
||||
31
tests/openbb_terminal/core/log/conftest.py
Normal file
31
tests/openbb_terminal/core/log/conftest.py
Normal file
@@ -0,0 +1,31 @@
|
||||
import pytest
|
||||
from openbb_terminal.core.log.generation.settings import (
|
||||
Settings,
|
||||
AppSettings,
|
||||
AWSSettings,
|
||||
LogSettings,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def settings(tmp_path):
|
||||
settings_obj = Settings(
|
||||
app_settings=AppSettings(
|
||||
commit_hash="MOCK_COMMIT_HASH",
|
||||
name="MOCK_NAME",
|
||||
identifier="MOCK_IDENTIFIER",
|
||||
session_id="MOCK_SESSION_ID",
|
||||
),
|
||||
aws_settings=AWSSettings(
|
||||
aws_access_key_id="MOCK_AWS_ACCESS_KEY_ID",
|
||||
aws_secret_access_key="MOCK_AWS_ACCESS_KEY", # pragma: allowlist secret
|
||||
),
|
||||
log_settings=LogSettings(
|
||||
directory=tmp_path,
|
||||
frequency="H",
|
||||
handler_list="file",
|
||||
rolling_clock=False,
|
||||
verbosity=20,
|
||||
),
|
||||
)
|
||||
return settings_obj
|
||||
@@ -0,0 +1,34 @@
|
||||
from openbb_terminal.core.log.generation.directories import get_log_dir, get_log_sub_dir
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def test_get_log_dir(mocker, tmp_path):
|
||||
mock_file_path = mocker.Mock()
|
||||
mock_file_path.parent.parent.parent.parent = tmp_path
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.generation.directories.Path",
|
||||
return_value=mock_file_path,
|
||||
)
|
||||
|
||||
log_dir = get_log_dir()
|
||||
|
||||
assert isinstance(log_dir, Path)
|
||||
assert log_dir.parent.parent == tmp_path
|
||||
|
||||
log_dir_bis = get_log_dir()
|
||||
|
||||
assert isinstance(log_dir, Path)
|
||||
assert log_dir.parent.parent == tmp_path
|
||||
assert log_dir_bis == log_dir
|
||||
|
||||
|
||||
def test_get_log_sub_dir(mocker, tmp_path):
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.generation.directories.get_log_dir",
|
||||
return_value=tmp_path,
|
||||
)
|
||||
name = "mock_sub_dir"
|
||||
sub_dir = get_log_sub_dir(name=name)
|
||||
|
||||
assert isinstance(sub_dir, Path)
|
||||
assert sub_dir.parent == tmp_path
|
||||
@@ -0,0 +1,44 @@
|
||||
import os
|
||||
from openbb_terminal.core.log.generation.expired_files import (
|
||||
get_timestamp_from_x_days,
|
||||
get_expired_file_list,
|
||||
remove_file_list,
|
||||
)
|
||||
|
||||
|
||||
def test_get_timestamp_from_x_days():
|
||||
timestamp = get_timestamp_from_x_days(x=1)
|
||||
|
||||
assert isinstance(timestamp, float)
|
||||
|
||||
|
||||
def test_get_expired_files(tmp_path):
|
||||
mock_file_list = [
|
||||
tmp_path.joinpath("mock_file_1"),
|
||||
tmp_path.joinpath("mock_file_2"),
|
||||
]
|
||||
|
||||
x = 2
|
||||
mock_timestamp = get_timestamp_from_x_days(x=x + 1)
|
||||
before_timestamp = get_timestamp_from_x_days(x=x)
|
||||
|
||||
for path in mock_file_list:
|
||||
path.touch()
|
||||
os.utime(path, times=(mock_timestamp, mock_timestamp))
|
||||
|
||||
expired_file_list = get_expired_file_list(
|
||||
directory=tmp_path, before_timestamp=before_timestamp
|
||||
)
|
||||
|
||||
assert len(expired_file_list) == len(mock_file_list)
|
||||
|
||||
for path in mock_file_list:
|
||||
assert path in expired_file_list
|
||||
|
||||
|
||||
def test_remove_file_list(mocker):
|
||||
file_list = [mocker.Mock(), mocker.Mock()]
|
||||
remove_file_list(file_list=file_list)
|
||||
|
||||
for file in file_list:
|
||||
file.unlink.assert_called_once()
|
||||
@@ -0,0 +1,77 @@
|
||||
import os
|
||||
import pytest
|
||||
from openbb_terminal.core.log.generation import formatter_with_exceptions as fwe
|
||||
from openbb_terminal.core.log.generation.settings import AppSettings
|
||||
from openbb_terminal.core.config.constants import REPO_DIR
|
||||
|
||||
app_settings = AppSettings(
|
||||
commit_hash="MOCK_COMMIT_HASH",
|
||||
name="MOCK_COMMIT_HASH",
|
||||
identifier="MOCK_COMMIT_HASH",
|
||||
session_id="MOCK_SESSION_ID",
|
||||
)
|
||||
|
||||
|
||||
formatter = fwe.FormatterWithExceptions(app_settings)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("exc, level", [(True, True), (False, "name"), (False, False)])
|
||||
def test_calculate_level_name(exc, level, mocker):
|
||||
mock = mocker.Mock()
|
||||
mock.exc_text = exc
|
||||
mock.levelname = level
|
||||
value = formatter.calculate_level_name(mock)
|
||||
assert value
|
||||
|
||||
|
||||
def test_extract_log_extra(mocker):
|
||||
mock = mocker.Mock()
|
||||
mock.exc_text = True
|
||||
mock.levelname = "name"
|
||||
value = formatter.extract_log_extra(mock)
|
||||
assert value
|
||||
|
||||
|
||||
def test_filter_piis():
|
||||
text = f"test 1.1.1.1 chavi@chavi.com {REPO_DIR.name}{os.sep} dd{os.sep}dd"
|
||||
value = formatter.filter_piis(text)
|
||||
assert value
|
||||
|
||||
|
||||
def test_filter_special_characters():
|
||||
text = "\nhello \this \rgreetings"
|
||||
value = formatter.filter_special_characters(text)
|
||||
assert value == " - hello his greetings"
|
||||
|
||||
|
||||
def test_detect_terminal_message():
|
||||
value = formatter.detect_terminal_message("The command doesn't exist on the")
|
||||
assert value or os.sep != "/"
|
||||
new_val = formatter.detect_terminal_message("boring message")
|
||||
assert new_val is False
|
||||
|
||||
|
||||
def test_filter_log_line():
|
||||
value = formatter.filter_log_line("stocks")
|
||||
assert value
|
||||
value = formatter.filter_log_line(
|
||||
"The command doesn't exist on the menu. - Traceback hello"
|
||||
)
|
||||
assert value
|
||||
|
||||
|
||||
def test_formatException_invalid():
|
||||
with pytest.raises(Exception):
|
||||
formatter.formatException(Exception("Big bad error"))
|
||||
|
||||
|
||||
def test_format(mocker):
|
||||
mock = mocker.Mock()
|
||||
mock.exc_text = "Text"
|
||||
mock.levelname = ""
|
||||
mock.created = 6
|
||||
mock.name = "Hello"
|
||||
mock.getMessage = lambda: "aeffew" or "3232432423"
|
||||
mock.stack_info = "info"
|
||||
value = formatter.format(mock)
|
||||
assert value
|
||||
@@ -0,0 +1,83 @@
|
||||
import random
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
import pytest
|
||||
from openbb_terminal.core.log.generation.path_tracking_file_handler import (
|
||||
PathTrackingFileHandler,
|
||||
)
|
||||
|
||||
|
||||
randint = random.randint(0, 999999999)
|
||||
|
||||
|
||||
def return_list(**_):
|
||||
return [Path("."), Path(".")]
|
||||
|
||||
|
||||
def test_build_log_file_path(settings):
|
||||
value = PathTrackingFileHandler.build_log_file_path(settings)
|
||||
assert value
|
||||
|
||||
|
||||
@pytest.mark.parametrize("start", [True, False])
|
||||
def test_build_log_sender(start, settings):
|
||||
value = PathTrackingFileHandler.build_log_sender(settings, start)
|
||||
assert value
|
||||
|
||||
|
||||
def test_clean_expired_files():
|
||||
PathTrackingFileHandler.clean_expired_files(datetime.now().timestamp())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("start, freq", [(True, "H"), (False, "M")])
|
||||
def test_build_rolling_clock(start, freq, mocker, settings):
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.generation.path_tracking_file_handler.LoggingClock"
|
||||
)
|
||||
handler = PathTrackingFileHandler(settings)
|
||||
value = PathTrackingFileHandler.build_rolling_clock(handler.doRollover, freq, start)
|
||||
assert value
|
||||
|
||||
|
||||
def test_build_rolling_clock_error(settings):
|
||||
handler = PathTrackingFileHandler(settings)
|
||||
with pytest.raises(AttributeError):
|
||||
PathTrackingFileHandler.build_rolling_clock(handler.doRollover, "T", True)
|
||||
|
||||
|
||||
def test_send_expired_files(mocker, settings):
|
||||
handler = PathTrackingFileHandler(settings)
|
||||
mocker.patch(
|
||||
"openbb_terminal.core.log.generation.path_tracking_file_handler.get_expired_file_list",
|
||||
return_list,
|
||||
)
|
||||
handler.send_expired_files(datetime.now().timestamp())
|
||||
|
||||
|
||||
def test_log_sender():
|
||||
value = PathTrackingFileHandler.log_sender
|
||||
assert value
|
||||
|
||||
|
||||
def test_rolling_clock():
|
||||
value = PathTrackingFileHandler.rolling_clock
|
||||
assert value
|
||||
|
||||
|
||||
def test_settings():
|
||||
value = PathTrackingFileHandler.settings
|
||||
assert value
|
||||
|
||||
|
||||
def test_doRollover(settings):
|
||||
handler = PathTrackingFileHandler(settings)
|
||||
handler.doRollover()
|
||||
|
||||
|
||||
def throw_error():
|
||||
raise Exception("Bad!")
|
||||
|
||||
|
||||
def test_close(settings):
|
||||
handler = PathTrackingFileHandler(settings)
|
||||
handler.close()
|
||||
107
tests/openbb_terminal/core/log/generation/test_settings.py
Normal file
107
tests/openbb_terminal/core/log/generation/test_settings.py
Normal file
@@ -0,0 +1,107 @@
|
||||
from openbb_terminal.core.log.generation.settings import (
|
||||
AppSettings,
|
||||
AWSSettings,
|
||||
LogSettings,
|
||||
Settings,
|
||||
)
|
||||
|
||||
|
||||
def test_aws_settings():
|
||||
aws_access_key_id = "MOCK_AWS_ACCESS_KEY_ID"
|
||||
aws_secret_access_key = "MOCK_AWS_SECRET_ACCESS_KEY"
|
||||
|
||||
aws_settings = AWSSettings(
|
||||
aws_access_key_id=aws_access_key_id,
|
||||
aws_secret_access_key=aws_secret_access_key,
|
||||
)
|
||||
|
||||
assert aws_settings.aws_access_key_id == aws_access_key_id
|
||||
assert aws_settings.aws_secret_access_key == aws_secret_access_key
|
||||
|
||||
|
||||
def test_app_settings():
|
||||
name = "MOCK_NAME"
|
||||
commit_hash = "MOCK_COMMIT_HASH"
|
||||
session_id = "MOCK_SESSION_ID"
|
||||
identifier = "MOCK_IDENTIFIER"
|
||||
|
||||
app_settings = AppSettings(
|
||||
name=name,
|
||||
commit_hash=commit_hash,
|
||||
session_id=session_id,
|
||||
identifier=identifier,
|
||||
)
|
||||
|
||||
assert app_settings.name == name
|
||||
assert app_settings.commit_hash == commit_hash
|
||||
assert app_settings.session_id == session_id
|
||||
assert app_settings.identifier == identifier
|
||||
|
||||
|
||||
def test_log_settings(tmp_path):
|
||||
directory = tmp_path
|
||||
frequency = "MOCK_FREQUENCY"
|
||||
handler_list = "MOCK_HANDLER_LIST"
|
||||
rolling_clock = "MOCK_ROLLING_CLOCK"
|
||||
verbosity = 20
|
||||
|
||||
log_settings = LogSettings(
|
||||
directory=directory,
|
||||
frequency=frequency,
|
||||
handler_list=handler_list,
|
||||
rolling_clock=rolling_clock,
|
||||
verbosity=verbosity,
|
||||
)
|
||||
|
||||
assert log_settings.directory == directory
|
||||
assert log_settings.frequency == frequency
|
||||
assert log_settings.handler_list == handler_list
|
||||
assert log_settings.rolling_clock == rolling_clock
|
||||
assert log_settings.verbosity == verbosity
|
||||
|
||||
|
||||
def test_settings(tmp_path):
|
||||
directory = tmp_path
|
||||
frequency = "MOCK_FREQUENCY"
|
||||
handler_list = "MOCK_HANDLER_LIST"
|
||||
rolling_clock = "MOCK_ROLLING_CLOCK"
|
||||
verbosity = 20
|
||||
log_settings = LogSettings(
|
||||
directory=directory,
|
||||
frequency=frequency,
|
||||
handler_list=handler_list,
|
||||
rolling_clock=rolling_clock,
|
||||
verbosity=verbosity,
|
||||
)
|
||||
|
||||
name = "MOCK_NAME"
|
||||
commit_hash = "MOCK_COMMIT_HASH"
|
||||
session_id = "MOCK_SESSION_ID"
|
||||
identifier = "MOCK_IDENTIFIER"
|
||||
app_settings = AppSettings(
|
||||
name=name,
|
||||
commit_hash=commit_hash,
|
||||
session_id=session_id,
|
||||
identifier=identifier,
|
||||
)
|
||||
|
||||
aws_access_key_id = "MOCK_AWS_ACCESS_KEY_ID"
|
||||
aws_secret_access_key = "MOCK_AWS_SECRET_ACCESS_KEY"
|
||||
aws_settings = AWSSettings(
|
||||
aws_access_key_id=aws_access_key_id,
|
||||
aws_secret_access_key=aws_secret_access_key,
|
||||
)
|
||||
|
||||
settings = Settings(
|
||||
app_settings=app_settings,
|
||||
aws_settings=aws_settings,
|
||||
log_settings=log_settings,
|
||||
)
|
||||
|
||||
assert settings.app_settings != app_settings
|
||||
assert settings.aws_settings != aws_settings
|
||||
assert settings.log_settings != log_settings
|
||||
|
||||
assert isinstance(settings.app_settings, AppSettings)
|
||||
assert isinstance(settings.aws_settings, AWSSettings)
|
||||
assert isinstance(settings.log_settings, LogSettings)
|
||||
5
tests/openbb_terminal/core/log/test_constants.py
Normal file
5
tests/openbb_terminal/core/log/test_constants.py
Normal file
@@ -0,0 +1,5 @@
|
||||
from openbb_terminal.core.log import constants
|
||||
|
||||
|
||||
def test_constants():
|
||||
assert constants.ARCHIVES_FOLDER_NAME
|
||||
@@ -10,6 +10,7 @@ from openbb_terminal.economy import cnn_model
|
||||
|
||||
@pytest.mark.default_cassette("test_get_feargreed_report")
|
||||
@pytest.mark.vcr
|
||||
@pytest.mark.linux
|
||||
@pytest.mark.parametrize(
|
||||
"indicator",
|
||||
[
|
||||
|
||||
100
tests/openbb_terminal/test_loggers.py
Normal file
100
tests/openbb_terminal/test_loggers.py
Normal file
@@ -0,0 +1,100 @@
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from openbb_terminal import loggers
|
||||
from openbb_terminal.core.log.generation.settings import (
|
||||
Settings,
|
||||
AppSettings,
|
||||
AWSSettings,
|
||||
LogSettings,
|
||||
)
|
||||
|
||||
settings = Settings(
|
||||
app_settings=AppSettings(
|
||||
commit_hash="MOCK_COMMIT_HASH",
|
||||
name="MOCK_COMMIT_HASH",
|
||||
identifier="MOCK_COMMIT_HASH",
|
||||
session_id="MOCK_SESSION_ID",
|
||||
),
|
||||
aws_settings=AWSSettings(
|
||||
aws_access_key_id="MOCK_AWS_ACCESS_KEY_ID",
|
||||
aws_secret_access_key="MOCK_AWS", # pragma: allowlist secret
|
||||
),
|
||||
log_settings=LogSettings(
|
||||
directory=Path("."),
|
||||
frequency="H",
|
||||
handler_list="file",
|
||||
rolling_clock=False,
|
||||
verbosity=20,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def throw_os_error():
|
||||
raise OSError("This is a test error")
|
||||
|
||||
|
||||
def throw_os_error_30():
|
||||
e = OSError()
|
||||
e.errno = 30
|
||||
raise e
|
||||
|
||||
|
||||
def throw_generic():
|
||||
raise Exception("This is a test error")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"to_mock", [None, throw_os_error, throw_os_error_30, throw_generic]
|
||||
)
|
||||
def test_get_app_id(to_mock, mocker):
|
||||
if to_mock:
|
||||
mocker.patch("openbb_terminal.loggers.get_log_dir", to_mock)
|
||||
with pytest.raises(Exception):
|
||||
value = loggers.get_app_id()
|
||||
else:
|
||||
value = loggers.get_app_id()
|
||||
assert value
|
||||
|
||||
|
||||
@pytest.mark.parametrize("git", [True, False])
|
||||
def test_get_commit_hash(mocker, git):
|
||||
mocker.patch("openbb_terminal.loggers.WITH_GIT", git)
|
||||
value = loggers.get_commit_hash()
|
||||
assert value
|
||||
|
||||
|
||||
def test_get_commit_hash_obff(mocker):
|
||||
mocker.patch("openbb_terminal.loggers.obbff")
|
||||
value = loggers.get_commit_hash()
|
||||
assert value
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger.")
|
||||
def test_add_stdout_handler():
|
||||
loggers.add_stdout_handler(settings)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger.")
|
||||
def test_add_stderr_handler():
|
||||
loggers.add_stderr_handler(settings)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger.")
|
||||
def test_add_nopp_handler():
|
||||
loggers.add_noop_handler(settings)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger + lead to file generation.")
|
||||
def test_add_file_handler():
|
||||
loggers.add_file_handler(settings)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger.")
|
||||
def test_setup_handlers():
|
||||
loggers.setup_handlers(settings)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Change the state of the logger.")
|
||||
def test_setup_logging(mocker):
|
||||
mocker.patch("openbb_terminal.loggers.setup_handlers")
|
||||
loggers.setup_logging(settings)
|
||||
Reference in New Issue
Block a user