(self, name: str, productIds: List[str])
| 236 | return res |
| 237 | |
| 238 | def create_ws_auth(self, name: str, productIds: List[str]): |
| 239 | subscribe = {} |
| 240 | timestamp = self.number_to_string(self.seconds()) |
| 241 | self.check_required_credentials() |
| 242 | isCloudAPiKey = (self.apiKey.find('organizations/') >= 0) or (self.secret.startswith('-----BEGIN')) |
| 243 | auth = timestamp + name + ','.join(productIds) |
| 244 | if not isCloudAPiKey: |
| 245 | subscribe['api_key'] = self.apiKey |
| 246 | subscribe['timestamp'] = timestamp |
| 247 | subscribe['signature'] = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256) |
| 248 | else: |
| 249 | if self.apiKey.startswith('-----BEGIN'): |
| 250 | raise ArgumentsRequired(self.id + ' apiKey should contain the name(eg: organizations/3b910e93....) and not the public key') |
| 251 | currentToken = self.safe_string(self.options, 'wsToken') |
| 252 | tokenTimestamp = self.safe_integer(self.options, 'wsTokenTimestamp', 0) |
| 253 | seconds = self.seconds() |
| 254 | if currentToken is None or tokenTimestamp + 120 < seconds: |
| 255 | # we should generate new token |
| 256 | token = self.create_auth_token(seconds) |
| 257 | self.options['wsToken'] = token |
| 258 | self.options['wsTokenTimestamp'] = seconds |
| 259 | subscribe['jwt'] = self.safe_string(self.options, 'wsToken') |
| 260 | return subscribe |
| 261 | |
| 262 | async def watch_ticker(self, symbol: str, params={}) -> Ticker: |
| 263 | """ |
no test coverage detected