v0.2.0

offish · Dec 29, 2024 · c3d7890 · c3d7890
1 parent e623476
commit c3d7890
Show file tree

Hide file tree

Showing 6 changed files with 91 additions and 77 deletions.
diff --git a/LICENSE b/LICENSE
@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2016 Michał Bukowski
+Copyright (c) 2016 Michał Bukowski [email protected]
 Copyright (c) 2024 offish
 
 Permission is hereby granted, free of charge, to any person obtaining a copy

diff --git a/README.md b/README.md
@@ -24,13 +24,19 @@ pip install steampie
 pip install --upgrade steampie
 ```
 
-## Usage
+## Differences from `steampy`
+<!-- * Added functionality for adding/removing friends -->
+* Removed market functionality due to Steam having strict ToS regarding market automation
+* Uses [ruff](https://github.com/astral-sh/ruff) for formatting and linting
+* Uses [pytest](https://pytest.org/) for unit tests
+
+<!-- ## Usage
 
 
 ## Examples
 
 
-## Developing
+## Developing -->
 
 
 ### Tests
@@ -50,7 +56,7 @@ make html
 ## License
 MIT License
 
-Copyright (c) 2016 [Michał Bukowski](mailto:[email protected])<br>
+Copyright (c) 2016 [Michał Bukowski](mailto:[email protected]) <br>
 Copyright (c) 2024 offish ([confern](https://steamcommunity.com/id/confern))
 
 Permission is hereby granted, free of charge, to any person obtaining a copy

diff --git a/conftest.py b/conftest.py
@@ -6,6 +6,20 @@
 from steampie.client import SteamClient
 
 
+def my_steam_id() -> str:
+    return "76561198253325712"
+
+
+@pytest.fixture
+def partner_steam_id() -> str:
+    return "76561198155938401"
+
+
+@pytest.fixture
+def partner_trade_url() -> str:
+    return "https://steamcommunity.com/tradeoffer/new/?partner=195672673&token=k3YfpXNU"
+
+
 @pytest.fixture
 def shared_secret() -> bytes:
     return b64encode(b"1234567890abcdefghij")
@@ -33,8 +47,9 @@ def credentials() -> dict:
 @pytest.fixture
 def steam_guard_file() -> dict:
     steam_guard_credentials = {}
+    steam_id = my_steam_id()
 
-    with open("./76561198253325712.maFile", "r") as f:
+    with open(f"./{steam_id}.maFile", "r") as f:
         steam_guard_credentials = json.load(f)
 
     steam_credentials = get_credentials()

diff --git a/src/steampie/__init__.py b/src/steampie/__init__.py
@@ -1,7 +1,7 @@
 # flake8: noqa
 __title__ = "steampie"
 __author__ = "offish"
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 __license__ = "MIT"
 
 from .client import SteamClient, SteamGuardLoginClient

diff --git a/src/steampie/client.py b/src/steampie/client.py
@@ -66,7 +66,8 @@ def set_proxies(self, proxies: dict) -> dict:
         if not isinstance(proxies, dict):
             raise TypeError(
                 "Proxy must be a dict. Example: "
-                r'\{"http": "http://login:password@host:port"\, "https": "http://login:password@host:port"\}',
+                + "{'http': 'http://login:password@host:port', "
+                + "'https': 'http://login:password@host:port'}",
             )
 
         if ping_proxy(proxies):
@@ -140,6 +141,7 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+        del exc_type, exc_val, exc_tb
         self.logout()
 
     @login_required
@@ -157,11 +159,12 @@ def api_call(
         params: dict | None = None,
     ) -> requests.Response:
         url = f"{SteamUrl.API_URL}/{interface}/{api_method}/{version}"
-        response = (
-            self._session.get(url, params=params)
-            if method == "GET"
-            else self._session.post(url, data=params)
-        )
+        response = None
+
+        if method == "GET":
+            response = self._session.get(url, params=params)
+        else:
+            response = self._session.post(url, data=params)
 
         if self.is_invalid_api_key(response):
             raise InvalidCredentials("Invalid API key")
@@ -216,16 +219,24 @@ def get_partner_inventory(
             else response_dict
         )
 
-    def _get_session_id(self) -> str:
-        return self._session.cookies.get_dict(
-            domain="steamcommunity.com", path="/"
-        ).get("sessionid")
+    def _get_session_id(self) -> str | None:
+        response = self._session.cookies.get_dict(domain="steamcommunity.com", path="/")
+        return response.get("sessionid")
 
     def get_trade_offers_summary(self) -> dict:
         params = {"key": self._api_key}
-        return self.api_call(
+        response = self.api_call(
             "GET", "IEconService", "GetTradeOffersSummary", "v1", params
-        ).json()
+        )
+        return response.json()
+
+    def _set_token_params(self, params: dict, use_webtoken: bool) -> dict:
+        if use_webtoken:
+            params["access_token"] = self._access_token
+        else:
+            params["key"] = self._api_key
+
+        return params
 
     def get_trade_offers(
         self,
@@ -236,9 +247,6 @@ def get_trade_offers(
         max_retry: int = 5,
     ) -> dict:
         params = {
-            "key" if not use_webtoken else "access_token": self._api_key
-            if not use_webtoken
-            else self._access_token,
             "get_sent_offers": int(get_sent_offers),
             "get_received_offers": int(get_received_offers),
             "get_descriptions": 1,
@@ -248,19 +256,22 @@ def get_trade_offers(
             "time_historical_cutoff": "",
         }
 
+        params = self._set_token_params(params, use_webtoken)
         response = self._try_to_get_trade_offers(params, max_retry)
+
         if response is None:
             raise ApiException("Cannot get proper json from get_trade_offers method")
+
         response_with_active_offers = self._filter_non_active_offers(response)
-        if merge:
-            return merge_items_with_descriptions_from_offers(
-                response_with_active_offers
-            )
-        else:
+
+        if not merge:
             return response_with_active_offers
 
+        return merge_items_with_descriptions_from_offers(response_with_active_offers)
+
     def _try_to_get_trade_offers(self, params: dict, max_retry: int) -> dict | None:
         response = None
+
         for _ in range(max_retry):
             try:
                 response = self.api_call(
@@ -269,7 +280,7 @@ def _try_to_get_trade_offers(self, params: dict, max_retry: int) -> dict | None:
                 break
             except json.decoder.JSONDecodeError:
                 time.sleep(2)
-                continue
+
         return response
 
     @staticmethod
@@ -296,10 +307,7 @@ def get_trade_offer(
         self, trade_offer_id: str, merge: bool = True, use_webtoken: bool = False
     ) -> dict:
         params = {"tradeofferid": trade_offer_id, "language": "english"}
-        if use_webtoken:
-            params["access_token"] = self._access_token
-        else:
-            params["key"] = self._api_key
+        params = self._set_token_params(params, use_webtoken)
 
         response = self.api_call(
             "GET", "IEconService", "GetTradeOffer", "v1", params
@@ -337,9 +345,8 @@ def get_trade_history(
             "include_failed": include_failed,
             "include_total": include_total,
         }
-        return self.api_call(
-            "GET", "IEconService", "GetTradeHistory", "v1", params
-        ).json()
+        response = self.api_call("GET", "IEconService", "GetTradeHistory", "v1", params)
+        return response.json()
 
     @login_required
     def get_trade_receipt(self, trade_id: str):
@@ -356,6 +363,7 @@ def accept_trade_offer(self, trade_offer_id: str) -> dict:
         trade_offer_state = TradeOfferState(
             trade["response"]["offer"]["trade_offer_state"]
         )
+
         if trade_offer_state is not TradeOfferState.Active:
             raise ApiException(
                 f"Invalid trade offer state: {trade_offer_state.name} ({trade_offer_state.value})"
@@ -374,6 +382,7 @@ def accept_trade_offer(self, trade_offer_id: str) -> dict:
         headers = {"Referer": self._get_trade_offer_url(trade_offer_id)}
 
         response = self._session.post(accept_url, data=params, headers=headers).json()
+
         if response.get("needs_mobile_confirmation", False):
             return self._confirm_transaction(trade_offer_id)
 
@@ -405,15 +414,13 @@ def _confirm_transaction(self, trade_offer_id: str) -> dict:
 
     def decline_trade_offer(self, trade_offer_id: str) -> dict:
         url = f"https://steamcommunity.com/tradeoffer/{trade_offer_id}/decline"
-        return self._session.post(
-            url, data={"sessionid": self._get_session_id()}
-        ).json()
+        response = self._session.post(url, data={"sessionid": self._get_session_id()})
+        return response.json()
 
     def cancel_trade_offer(self, trade_offer_id: str) -> dict:
         url = f"https://steamcommunity.com/tradeoffer/{trade_offer_id}/cancel"
-        return self._session.post(
-            url, data={"sessionid": self._get_session_id()}
-        ).json()
+        response = self._session.post(url, data={"sessionid": self._get_session_id()})
+        return response.json()
 
     @login_required
     def make_offer(
@@ -536,6 +543,7 @@ def make_offer_with_url(
         }
 
         response = self._session.post(url, data=params, headers=headers).json()
+
         if confirm_trade and response.get("needs_mobile_confirmation"):
             response.update(self._confirm_transaction(response["tradeofferid"]))
 
@@ -563,7 +571,6 @@ def _friend_ajax_request(self, method: str, steam_id: str, accept: int = 0) -> d
                 f"Request failed with status code {response.status_code}"
             )
 
-        # returns True or json object if request was succes, otherwise False
         return response.json()
 
     @login_required

diff --git a/tests/test_client.py b/tests/test_client.py
@@ -67,8 +67,25 @@ def test_get_trade_offers(client: SteamClient) -> None:
 #     assert response_dict["response"] == {}
 
 
-def test_make_offer(client: SteamClient) -> None:
-    partner_steam_id = "76561198449257208"
+def test_make_offer_url(
+    client: SteamClient, partner_steam_id: str, partner_trade_url: str
+) -> None:
+    game = GameOptions.TF2
+    my_items = client.get_my_inventory(GameOptions.TF2)
+    partner_items = client.get_partner_inventory(partner_steam_id, game)
+    my_first_item = next(iter(my_items.values()))
+    partner_first_item = next(iter(partner_items.values()))
+    my_asset = Asset(my_first_item["id"], game)
+    partner_asset = Asset(partner_first_item["id"], game)
+    response = client.make_offer_with_url(
+        [my_asset], [partner_asset], partner_trade_url, "test offer"
+    )
+
+    assert response is not None
+    assert "tradeofferid" in response
+
+
+def test_make_offer(client: SteamClient, partner_steam_id: str) -> None:
     game = GameOptions.TF2
     my_items = client.get_my_inventory(GameOptions.TF2)
     partner_items = client.get_partner_inventory(partner_steam_id, game)
@@ -84,40 +101,9 @@ def test_make_offer(client: SteamClient) -> None:
     assert "tradeofferid" in response
 
 
-# def test_make_offer_url() -> None:
-#     partner_account_id = "488991480"
-#     partner_token = "7vqRtBpC"
-#     sample_trade_url = f"https://steamcommunity.com/tradeoffer/new/?partner={partner_account_id}&token={partner_token}"
-#     client = SteamClient(self.credentials.api_key)
-#     client.login(
-#         self.credentials.login, self.credentials.password, self.steam_guard_file
-#     )
-#     client._session.request("HEAD", "http://steamcommunity.com")
-#     partner_steam_id = account_id_to_steam_id(partner_account_id)
-#     game = GameOptions.CS
-#     my_items = client.get_my_inventory(game, merge=False)["rgInventory"]
-#     partner_items = client.get_partner_inventory(partner_steam_id, game, merge=False)[
-#         "rgInventory"
-#     ]
-#     my_first_item = next(iter(my_items.values()))
-#     partner_first_item = next(iter(partner_items.values()))
-#     my_asset = Asset(my_first_item["id"], game)
-#     partner_asset = Asset(partner_first_item["id"], game)
-#     response = client.make_offer_with_url(
-#         [my_asset], [partner_asset], sample_trade_url, "TESTOWA OFERTA"
-#     )
-
-#     assert response is not None
-#     assert "tradeofferid" in response
-
-
-# def test_get_escrow_duration(client:SteamClient) -> None:
-#     # A sample trade URL with escrow time of 15 days cause mobile auth not added
-#     sample_trade_url = (
-#         "https://steamcommunity.com/tradeoffer/new/?partner=314218906&token=sgA4FdNm"
-#     )
-#     response = client.get_escrow_duration(sample_trade_url)
-#     assert response == 15
+def test_get_escrow_duration(client: SteamClient, partner_trade_url: str) -> None:
+    response = client.get_escrow_duration(partner_trade_url)
+    assert response == 0
 
 
 def test_logout(client: SteamClient) -> None: