Minor code fixes. Major test refactoring. #3 Complete unit testing ma…

…in methods/functions.

mkdocs.yml

@@ -4,8 +4,6 @@ site_url: https://somespecialone.github.io/python-steam-tradeoffer-manager/
 repo_name: somespecialone/python-steam-tradeoffer-manager
 repo_url: https://github.com/somespecialone/python-steam-tradeoffer-manager/
 
-edit_uri: edit/dev/docs/
-
 copyright: Copyright &copy; 2022 <a href="https://github.com/somespecialone" target="_blank" rel="noopener">Dmitriy Tkachenko</a>
 
 theme:

pyproject.toml

@@ -39,7 +39,7 @@ testpaths = ["tests"]
 
 [tool.coverage.run]
 source = ["steam_tradeoffer_manager"]
-omit = ["*/_monkey_patch.py", "*/__init__.py"]
+omit = ["*/_monkey_patch.py", "*/__init__.py", "*/abc.py"]
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

steam_tradeoffer_manager/base/bot.py

@@ -155,7 +155,7 @@ async def restart(self) -> None:
         await self.start()
 
     @property
-    def randomizer(self) -> Callable[[], int] | None:
+    def randomizer(self) -> Callable[[...], int] | None:
         try:
             return self._randomizer or self.pool.randomizer
         except AttributeError:  # if bot don't bounded to pool

steam_tradeoffer_manager/base/enums.py

@@ -15,11 +15,11 @@ class BotState(enum.Enum):
 
 class Timings:
     @staticmethod
-    def _randomizer(*_range: int):
-        return lambda: round(randint(*_range) * 60 * 60 + (random() * 60 * 60))
+    def _randomizer(*_range: int) -> Callable[[...], int]:
+        return lambda *a, **ka: round(randint(*_range) * 60 * 60 + (random() * 60 * 60))
 
     @classmethod
-    def custom(cls, from_: int, to: int) -> Callable[[], int]:
+    def custom(cls, from_: int, to: int) -> Callable[[...], int]:
         return cls._randomizer(from_, to)
 
     HOUR = _randomizer(0, 1)  # not recommended

steam_tradeoffer_manager/base/pool.py

@@ -7,19 +7,19 @@
 from .mixins import PoolBotMixin
 from .enums import ONCE_EVERY
 
-__all__ = ('SteamBotPool',)
+__all__ = ("SteamBotPool",)
 
 _log = logging.getLogger(__name__)
-_B = TypeVar('_B', bound="_bot.SteamBot")
-_I = TypeVar('_I')
-_D = TypeVar('_D')  # not sure if this is working
+_B = TypeVar("_B", bound="_bot.SteamBot")
+_I = TypeVar("_I")
+_D = TypeVar("_D")  # not sure if this is working
 
 
 class SteamBotPool(Generic[_I, _B], AbstractBasePool):
     """Steam bots pool"""
 
     # randomize time to sleep between restarting bots in pool instances of this class
-    randomizer: Callable[[], int] | None = ONCE_EVERY.FOUR_HOURS
+    randomizer: Callable[[...], int] | None = ONCE_EVERY.FOUR_HOURS
     whitelist: set[int] | None = None  # whitelist with steam id's of admins/owners/etc
     domain: str = "steam.py"  # domain to register new api key
 
@@ -38,6 +38,8 @@ def add(self, bot: _B, raise_=True) -> None:
         Add bot instance to pool
         :param bot: bot instance
         :param raise_: raise `ConstraintException` if bot in this pool or bounded to other pool. Default - `True`
+        :return `None`
+        :raises ConstraintException
         """
         if bot.pool:
             import warnings

steam_tradeoffer_manager/bot.py

@@ -82,6 +82,7 @@ def __init__(
             password: str,
             shared_secret: str,
             identity_secret: str,
+            *,
             offer_cancel_delay: timedelta | None = None,
             prefetch_games: tuple[SteamGame] = (),
             **kwargs
@@ -233,17 +234,16 @@ async def send_offer(self, offer: ManagerTradeOffer) -> None:
 
     async def on_trade_receive(self, trade: steam.TradeOffer) -> None:
         """
-        Fetches and save inventories when received trade offer
+        Fetches and save inventories when receives trade offer from user in whitelist.
         """
         if trade.partner.id64 in (self.whitelist or ()):
             await trade.accept()
             fetched = set()
             for item in trade.items_to_receive:
-                if item.game().name not in fetched and item.game().id:
-                    await self.inventory.fetch_game_inventory(item.game())
-                    fetched.add(item.game().name or item.game().id)
+                if item.game.name not in fetched and item.game.id not in fetched:
+                    await self.inventory.fetch_game_inventory(item.game)
+                    fetched.add(item.game.name or item.game.id)
 
-    # TODO: there or somewhere after dispatch needs to clear trades from this closed trade,
     def _close_trade_offer(self, trade: steam.TradeOffer):
         if trade.is_our_offer():  # there safe to call is_our_offer
             if trade.id in self.manager_trades:

steam_tradeoffer_manager/inventory.py

@@ -1,4 +1,4 @@
-from typing import Iterator, TypeVar, Generic, TypeAlias, overload
+from typing import Iterator, TypeVar, Generic, TypeAlias
 from collections.abc import MutableMapping
 from weakref import WeakValueDictionary
 
@@ -13,25 +13,24 @@
 SteamGame: TypeAlias = "Game | StatefulGame"
 BotInventory: TypeAlias = "BaseInventory[BotItem[_B]]"
 AssetId: TypeAlias = "int"
-AnyType: TypeAlias = "Game | StatefulGame | int"
 
 
 class GamesInventory(MutableMapping[AssetId, BotItem[_B]], Generic[_B]):
     """Container class created to store fetched inventories"""
 
-    __slots__ = ('_storage', 'owner', "_inventory_storage")
+    __slots__ = ('_items_storage', 'owner', "_inventories_storage")
 
     def __init__(self, owner: _B):
-        self._inventory_storage: dict[int | str, BotInventory] = {}  # default inventories with refs to items
-        self._storage: WeakValueDictionary[AssetId, BotItem[_B]] = WeakValueDictionary()  # weak refs to items
+        self._inventories_storage: dict[int, BotInventory] = {}  # default inventories with refs to items
+        self._items_storage: WeakValueDictionary[AssetId, BotItem[_B]] = WeakValueDictionary()  # weak refs to items
         self.owner = owner
 
     async def update_all(self) -> None:
         """Update all saved inventories"""
-        for inv in self._inventory_storage.values():
+        for inv in self._inventories_storage.values():
             await inv.update()
             inv.items = tuple(map(lambda i: BotItem(i, self.owner), inv.items))
-            self._storage.update({bot_item.asset_id: bot_item for bot_item in inv.items})
+            self._items_storage.update({bot_item.asset_id: bot_item for bot_item in inv.items})
 
             self.owner.dispatch_to_manager("inventory_update", inv)
 
@@ -49,54 +48,46 @@ def game_inventories(self) -> list[BotInventory]:
         Get all game inventories of this bot.
         :return: list[`BotInventory`]
         """
-        return list(iter(self._inventory_storage.values()))
-
-    @overload
-    def get(self, game: SteamGame, default: _D = None) -> BotInventory | _D:
-        """Get cached bot inventory"""
+        return list(iter(self._inventories_storage.values()))
 
     def get(self, asset_id: AssetId, default: _D = None) -> BotItem[_B] | _D:
-        """Get cached bot item"""
-        return self._inventory_storage.get(asset_id.name or asset_id.id, default) if isinstance(asset_id, Game)\
-            else self._storage.get(asset_id, default)
+        """Get cached bot item."""
+        return self._items_storage.get(asset_id, default)
+
+    def get_game_inventory(self, game: SteamGame, default: _D = None) -> BotInventory | _D:
+        """Get inventory for given game."""
+        return self._inventories_storage.get(game.id, default)
 
     async def fetch_game_inventory(self, game: SteamGame) -> BotInventory:
-        """Fetch inventory from steam servers and cache it"""
-        inv = await self.owner.user.inventory(game)  # again type hinting
+        """Fetch inventory from steam servers and cache it."""
+        inv: BaseInventory = await self.owner.user.inventory(game)  # again type hinting :(
         inv.items = tuple(map(lambda i: BotItem(i, self.owner), inv.items))
-        self._inventory_storage[game.name or game.id] = inv
-        self._storage.update({bot_item.asset_id: bot_item for bot_item in inv.items})
+        self._inventories_storage[game.id] = inv
+        self._items_storage.update({bot_item.asset_id: bot_item for bot_item in inv.items})
 
         self.owner.dispatch_to_manager("inventory_update", inv)
 
         return inv
 
-    @overload
-    def pop(self, game: SteamGame) -> BotInventory: ...
-
-    def pop(self, asset_id: AssetId) -> BotItem[_B]:
-        return self._inventory_storage.pop(asset_id.name or asset_id.id) if isinstance(asset_id, Game) \
-            else self._storage.pop(asset_id)
-
-    # mapping methods
+    # mapping methods only for items
 
     def __iter__(self) -> Iterator[BotItem[_B]]:
-        return iter(self._storage.values())
+        return iter(self._items_storage.values())
 
     def __len__(self) -> int:
-        return len(self._storage)
+        return len(self._items_storage)
 
-    def __getitem__(self, k: AnyType) -> BotItem[_B]:
-        return self._inventory_storage[k.name or k.id] if isinstance(k, Game) else self._storage[k]
+    def __getitem__(self, k: AssetId) -> BotItem[_B]:
+        return self._items_storage[k]
 
     def __delitem__(self, v: AssetId) -> None:
-        del self._storage[v]
+        del self._items_storage[v]
 
     def __setitem__(self, k: AssetId, v: BotItem[_B]) -> None:
-        self._storage[k] = v
+        self._items_storage[k] = v
 
-    def __contains__(self, k: AnyType) -> bool:
-        return k.name or k.id in self._inventory_storage if isinstance(k, Game) else k in self._storage
+    def __contains__(self, k: AssetId) -> bool:
+        return k in self._items_storage
 
 
 from . import bot

steam_tradeoffer_manager/items.py

@@ -16,7 +16,7 @@
 # extremely rarely asset id can be not unique
 class ManagerItems(MutableMapping[AssetId, BotItem[_B]], Generic[_M, _B]):
     """Items storage for TradeOfferManager.
-    Contain all items from `ManagerBot` bounded to `owner` manager"""
+    Contain all items from `ManagerBot`s bounded to `owner` manager"""
 
     def __init__(self, owner: _M):
         self.owner = owner
@@ -28,9 +28,6 @@ def add(self, item: BotItem):
     def get(self, k: AssetId, _default: _D = None) -> BotItem | _D:
         return self._storage.get(k, _default)
 
-    def pop(self, id: AssetId) -> BotItem:
-        return self._storage.pop(id)
-
     def __setitem__(self, k: AssetId, v: BotItem) -> None:
         if k == v.asset_id:
             raise ValueError("Key and asset id must be the same")

steam_tradeoffer_manager/mixins.py

@@ -19,6 +19,7 @@ class ManagerDispatchMixin:
     def errors(self) -> list[Exception]:
         return getattr(self, "_errors", [])
 
+    # TODO: i sure i forgot smth there
     async def on_error(self, event: str, error: Exception, *args, **kwargs):
         if errors := getattr(self, "_errors", None):
             errors.append(error)

steam_tradeoffer_manager/trades.py

@@ -46,11 +46,9 @@ def get(self, k: TradeOfferId, _default: _D = None) -> ManagerTradeOffer | _D:
         return self._storage.get(k, _default)
 
     def pop(self, id: TradeOfferId) -> ManagerTradeOffer:
-        return self._storage.pop(id)
-
-    @property
-    def storage(self):
-        return self._storage
+        offer = self[id]
+        del self[id]
+        return offer
 
     def __setitem__(self, k: TradeOfferId, v: ManagerTradeOffer) -> None:
         if not v.id:
@@ -60,6 +58,8 @@ def __setitem__(self, k: TradeOfferId, v: ManagerTradeOffer) -> None:
         self._storage[k] = v
 
     def __delitem__(self, v: TradeOfferId) -> None:
+        if self[v].is_active:
+            raise TypeError("You can't delete active offer!")
         del self._storage[v]
 
     def __getitem__(self, k: TradeOfferId) -> ManagerTradeOffer:

tests/conftest.py

@@ -4,8 +4,8 @@
 import pytest
 
 from mocks import *
-from const_data import *
-from steam_tradeoffer_manager import ManagerBot
+from data import *
+from steam_tradeoffer_manager import ManagerBot, TradeOfferManager
 
 
 @pytest.fixture(scope="session")
@@ -21,13 +21,7 @@ def event_loop():
 
 @pytest.fixture(scope="class")
 async def bot(event_loop):
-    bot_instance = ManagerBot(**{
-        "username": BOT_USERNAME,
-        "password": BOT_PASSWORD,
-        "shared_secret": SHARED_SECRET,
-        "identity_secret": IDENTITY_SECRET,
-        "id": BOT_ID
-    })
+    bot_instance = ManagerBot(**bot_data())
     bot_instance.loop = event_loop
 
     yield bot_instance