CAT_TOM/pyasic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: Update unittests.

Browse Source
This commit is contained in:
UpstreamData
2023-02-23 11:42:20 -07:00
parent 5ff0074d33
commit 2dc4b2cb8a
5 changed files with 227 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
tests/api_tests/__init__.py Normal file
View File

@@ -0,0 +1,211 @@
# ------------------------------------------------------------------------------
# Copyright 2022 Upstream Data Inc -
# -
# Licensed under the Apache License, Version 2.0 (the "License"); -
# you may not use this file except in compliance with the License. -
# You may obtain a copy of the License at -
# -
# http://www.apache.org/licenses/LICENSE-2.0 -
# -
# Unless required by applicable law or agreed to in writing, software -
# distributed under the License is distributed on an "AS IS" BASIS, -
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
import asyncio
import unittest
from unittest.mock import MagicMock, Mock, patch
from pyasic.API.bmminer import BMMinerAPI
from pyasic.API.bosminer import BOSMinerAPI
from pyasic.API.btminer import BTMinerAPI
from pyasic.API.cgminer import CGMinerAPI
class TestBMMinerAPI(unittest.IsolatedAsyncioTestCase):
async def create_mock_connection(self, reader_data: bytes = b""):
mock_reader = Mock(asyncio.StreamReader)
mock_reader.read.side_effect = [reader_data, None]
mock_writer = Mock(asyncio.StreamWriter)
mock_writer.write.return_value = None
mock_writer.drain.return_value = None
mock_writer.get_extra_info.return_value = (self.ip, self.port)
mock_connection = MagicMock()
mock_connection.__aenter__.return_value = (mock_reader, mock_writer)
mock_connection.__aexit__.return_value = None
return mock_connection
def setUp(self):
self.ip = "192.168.0.1"
self.port = 4028
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_send_command(self, mock_send_bytes):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
mock_send_bytes.return_value = b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BMMiner"}], "SUMMARY":[{}], "id": 1}'
response = await miner_api.send_command("summary")
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_all_commands(self, mock_send_bytes):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
for command in miner_api.commands:
mock_send_bytes.return_value = (
b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BMMiner"}], "'
+ command.upper().encode("utf-8")
+ b'":[{}], "id": 1}'
)
try:
command_func = getattr(miner_api, command)
except AttributeError:
pass
else:
try:
response = await command_func()
except TypeError:
# probably addpool or something
pass
else:
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
async def test_init(self):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
self.assertEqual(str(miner_api.ip), self.ip)
self.assertEqual(miner_api.port, self.port)
self.assertEqual(str(miner_api), "BMMinerAPI: 192.168.0.1")
class TestBTMinerAPI(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.ip = "192.168.0.1"
self.port = 4028
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_send_command(self, mock_send_bytes):
mock_send_bytes.return_value = b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BTMiner"}], "SUMMARY":[{"sumamry": 1}], "id": 1}'
miner_api = BTMinerAPI(ip=self.ip, port=self.port)
response = await miner_api.send_command("summary")
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_all_commands(self, mock_send_bytes):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
for command in miner_api.commands:
mock_send_bytes.return_value = (
b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BTMiner"}], "'
+ command.upper().encode("utf-8")
+ b'":[{}], "id": 1}'
)
try:
command_func = getattr(miner_api, command)
except AttributeError:
pass
else:
try:
response = await command_func()
except TypeError:
# probably addpool or something
pass
else:
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
async def test_init(self):
miner_api = BTMinerAPI(ip=self.ip, port=self.port)
self.assertEqual(str(miner_api.ip), self.ip)
self.assertEqual(miner_api.port, self.port)
self.assertEqual(str(miner_api), "BTMinerAPI: 192.168.0.1")
class TestCGMinerAPI(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.ip = "192.168.0.1"
self.port = 4028
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_send_command(self, mock_send_bytes):
mock_send_bytes.return_value = b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"CGMiner"}], "SUMMARY":[{"sumamry": 1}], "id": 1}'
miner_api = CGMinerAPI(ip=self.ip, port=self.port)
response = await miner_api.send_command("summary")
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_all_commands(self, mock_send_bytes):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
for command in miner_api.commands:
mock_send_bytes.return_value = (
b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"CGMiner"}], "'
+ command.upper().encode("utf-8")
+ b'":[{}], "id": 1}'
)
try:
command_func = getattr(miner_api, command)
except AttributeError:
pass
else:
try:
response = await command_func()
except TypeError:
# probably addpool or something
pass
else:
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
async def test_init(self):
miner_api = CGMinerAPI(ip=self.ip, port=self.port)
self.assertEqual(str(miner_api.ip), self.ip)
self.assertEqual(miner_api.port, self.port)
self.assertEqual(str(miner_api), "CGMinerAPI: 192.168.0.1")
class TestBOSMinerAPI(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.ip = "192.168.0.1"
self.port = 4028
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_send_command(self, mock_send_bytes):
mock_send_bytes.return_value = b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BOSMiner"}], "SUMMARY":[{"sumamry": 1}], "id": 1}'
miner_api = BOSMinerAPI(ip=self.ip, port=self.port)
response = await miner_api.send_command("summary")
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
@patch("pyasic.API.BaseMinerAPI._send_bytes")
async def test_all_commands(self, mock_send_bytes):
miner_api = BMMinerAPI(ip=self.ip, port=self.port)
for command in miner_api.commands:
mock_send_bytes.return_value = (
b'{"STATUS":[{"STATUS":"S","When":1618486231,"Code":7,"Msg":"BOSMiner"}], "'
+ command.upper().encode("utf-8")
+ b'":[{}], "id": 1}'
)
try:
command_func = getattr(miner_api, command)
except AttributeError:
pass
else:
try:
response = await command_func()
except TypeError:
# probably addpool or something
pass
else:
self.assertIsInstance(response, dict)
self.assertEqual(response["STATUS"][0]["STATUS"], "S")
async def test_init(self):
miner_api = BOSMinerAPI(ip=self.ip, port=self.port)
self.assertEqual(str(miner_api.ip), self.ip)
self.assertEqual(miner_api.port, self.port)
self.assertEqual(str(miner_api), "BOSMinerAPI: 192.168.0.1")
if __name__ == "__main__":
unittest.main()

10
tests/miners_tests/__init__.py
View File

@@ -17,6 +17,7 @@ import asyncio
import inspect
import sys
import unittest
import warnings
from pyasic.miners._backends import CGMiner # noqa
from pyasic.miners.base import BaseMiner
@@ -25,6 +26,7 @@ from pyasic.miners.miner_factory import MINER_CLASSES, MinerFactory
class MinersTest(unittest.TestCase):
def test_miner_model_creation(self):
warnings.filterwarnings("ignore")
for miner_model in MINER_CLASSES.keys():
for miner_api in MINER_CLASSES[miner_model].keys():
with self.subTest(
@@ -38,6 +40,8 @@ class MinersTest(unittest.TestCase):
)
def test_miner_backend_backup_creation(self):
warnings.filterwarnings("ignore")
backends = inspect.getmembers(
sys.modules["pyasic.miners._backends"], inspect.isclass
)
@@ -48,6 +52,8 @@ class MinersTest(unittest.TestCase):
self.assertTrue(isinstance(miner, miner_class))
def test_miner_type_creation_failure(self):
warnings.filterwarnings("ignore")
backends = inspect.getmembers(
sys.modules["pyasic.miners._types"], inspect.isclass
)
@@ -70,6 +76,8 @@ class MinersTest(unittest.TestCase):
class MinerFactoryTest(unittest.TestCase):
def test_miner_factory_creation(self):
warnings.filterwarnings("ignore")
self.assertDictEqual(MinerFactory().miners, {})
miner_factory = MinerFactory()
self.assertIs(MinerFactory(), miner_factory)
@@ -86,6 +94,8 @@ class MinerFactoryTest(unittest.TestCase):
self.assertListEqual(_miners, [])
def test_miner_selection(self):
warnings.filterwarnings("ignore")
for miner_model in MINER_CLASSES.keys():
with self.subTest():
miner = MinerFactory()._select_miner_from_classes(