jxtech /
wechatpy
| Total Complexity | 234 |
| Total Lines | 917 |
| Duplicated Lines | 1.09 % |
| Changes | 0 | ||
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like test_client often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | # -*- coding: utf-8 -*- |
||
| 2 | from __future__ import absolute_import, print_function, unicode_literals |
||
| 3 | |||
| 4 | import os |
||
| 5 | import unittest |
||
| 6 | from datetime import datetime |
||
| 7 | |||
| 8 | import six |
||
| 9 | from httmock import HTTMock, response, urlmatch |
||
| 10 | |||
| 11 | from wechatpy import WeChatClient |
||
| 12 | from wechatpy.exceptions import WeChatClientException |
||
| 13 | from wechatpy.utils import json |
||
| 14 | |||
| 15 | _TESTS_PATH = os.path.abspath(os.path.dirname(__file__)) |
||
| 16 | _FIXTURE_PATH = os.path.join(_TESTS_PATH, 'fixtures') |
||
| 17 | |||
| 18 | |||
| 19 | @urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$') |
||
| 20 | def wechat_api_mock(url, request): |
||
| 21 | path = url.path.replace('/cgi-bin/', '').replace('/', '_') |
||
| 22 | if path.startswith('_'): |
||
| 23 | path = path[1:] |
||
| 24 | res_file = os.path.join(_FIXTURE_PATH, '%s.json' % path) |
||
| 25 | content = { |
||
| 26 | 'errcode': 99999, |
||
| 27 | 'errmsg': 'can not find fixture %s' % res_file, |
||
| 28 | } |
||
| 29 | headers = { |
||
| 30 | 'Content-Type': 'application/json' |
||
| 31 | } |
||
| 32 | try: |
||
| 33 | with open(res_file, 'rb') as f: |
||
| 34 | content = json.loads(f.read().decode('utf-8')) |
||
| 35 | except (IOError, ValueError) as e: |
||
| 36 | content['errmsg'] = 'Loads fixture {0} failed, error: {1}'.format( |
||
| 37 | res_file, |
||
| 38 | e |
||
| 39 | ) |
||
| 40 | return response(200, content, headers, request=request) |
||
| 41 | |||
| 42 | |||
| 43 | class WeChatClientTestCase(unittest.TestCase): |
||
| 44 | app_id = '123456' |
||
| 45 | secret = '123456' |
||
| 46 | |||
| 47 | def setUp(self): |
||
| 48 | self.client = WeChatClient(self.app_id, self.secret) |
||
| 49 | |||
| 50 | def test_two_client_not_equal(self): |
||
| 51 | client2 = WeChatClient('654321', '654321', '987654321') |
||
| 52 | assert self.client != client2 |
||
| 53 | assert self.client.user != client2.user |
||
| 54 | assert id(self.client.menu) != id(client2.menu) |
||
| 55 | with HTTMock(wechat_api_mock): |
||
| 56 | self.client.fetch_access_token() |
||
| 57 | assert self.client.access_token != client2.access_token |
||
| 58 | |||
| 59 | def test_subclass_client_ok(self): |
||
| 60 | class TestClient(WeChatClient): |
||
| 61 | pass |
||
| 62 | |||
| 63 | client = TestClient('12345', '123456', '123456789') |
||
| 64 | self.assertEqual(client, client.user._client) |
||
| 65 | |||
| 66 | def test_fetch_access_token(self): |
||
| 67 | with HTTMock(wechat_api_mock): |
||
| 68 | token = self.client.fetch_access_token() |
||
| 69 | self.assertEqual('1234567890', token['access_token']) |
||
| 70 | self.assertEqual(7200, token['expires_in']) |
||
| 71 | self.assertEqual('1234567890', self.client.access_token) |
||
| 72 | |||
| 73 | def test_upload_media(self): |
||
| 74 | media_file = six.StringIO('nothing') |
||
| 75 | with HTTMock(wechat_api_mock): |
||
| 76 | media = self.client.media.upload('image', media_file) |
||
| 77 | self.assertEqual('image', media['type']) |
||
| 78 | self.assertEqual('12345678', media['media_id']) |
||
| 79 | |||
| 80 | def test_user_get_group_id(self): |
||
| 81 | with HTTMock(wechat_api_mock): |
||
| 82 | group_id = self.client.user.get_group_id('123456') |
||
| 83 | self.assertEqual(102, group_id) |
||
| 84 | |||
| 85 | def test_create_group(self): |
||
| 86 | with HTTMock(wechat_api_mock): |
||
| 87 | group = self.client.group.create('test') |
||
| 88 | self.assertEqual(1, group['group']['id']) |
||
| 89 | self.assertEqual('test', group['group']['name']) |
||
| 90 | |||
| 91 | def test_group_get(self): |
||
| 92 | with HTTMock(wechat_api_mock): |
||
| 93 | groups = self.client.group.get() |
||
| 94 | self.assertEqual(5, len(groups)) |
||
| 95 | |||
| 96 | def test_group_getid(self): |
||
| 97 | with HTTMock(wechat_api_mock): |
||
| 98 | group = self.client.group.get('123456') |
||
| 99 | self.assertEqual(102, group) |
||
| 100 | |||
| 101 | def test_group_update(self): |
||
| 102 | with HTTMock(wechat_api_mock): |
||
| 103 | result = self.client.group.update(102, 'test') |
||
| 104 | self.assertEqual(0, result['errcode']) |
||
| 105 | |||
| 106 | def test_group_move_user(self): |
||
| 107 | with HTTMock(wechat_api_mock): |
||
| 108 | result = self.client.group.move_user('test', 102) |
||
| 109 | self.assertEqual(0, result['errcode']) |
||
| 110 | |||
| 111 | def test_group_delete(self): |
||
| 112 | with HTTMock(wechat_api_mock): |
||
| 113 | result = self.client.group.delete(123456) |
||
| 114 | self.assertEqual(0, result['errcode']) |
||
| 115 | |||
| 116 | def test_send_text_message(self): |
||
| 117 | with HTTMock(wechat_api_mock): |
||
| 118 | result = self.client.message.send_text(1, 'test', account='test') |
||
| 119 | self.assertEqual(0, result['errcode']) |
||
| 120 | |||
| 121 | def test_send_image_message(self): |
||
| 122 | with HTTMock(wechat_api_mock): |
||
| 123 | result = self.client.message.send_image(1, '123456') |
||
| 124 | self.assertEqual(0, result['errcode']) |
||
| 125 | |||
| 126 | def test_send_voice_message(self): |
||
| 127 | with HTTMock(wechat_api_mock): |
||
| 128 | result = self.client.message.send_voice(1, '123456') |
||
| 129 | self.assertEqual(0, result['errcode']) |
||
| 130 | |||
| 131 | def test_send_video_message(self): |
||
| 132 | with HTTMock(wechat_api_mock): |
||
| 133 | result = self.client.message.send_video( |
||
| 134 | 1, '123456', 'test', 'test' |
||
| 135 | ) |
||
| 136 | self.assertEqual(0, result['errcode']) |
||
| 137 | |||
| 138 | def test_send_music_message(self): |
||
| 139 | with HTTMock(wechat_api_mock): |
||
| 140 | result = self.client.message.send_music( |
||
| 141 | 1, 'http://www.qq.com', 'http://www.qq.com', |
||
| 142 | '123456', 'test', 'test' |
||
| 143 | ) |
||
| 144 | self.assertEqual(0, result['errcode']) |
||
| 145 | |||
| 146 | def test_send_articles_message(self): |
||
| 147 | with HTTMock(wechat_api_mock): |
||
| 148 | articles = [{ |
||
| 149 | 'title': 'test', |
||
| 150 | 'description': 'test', |
||
| 151 | 'url': 'http://www.qq.com', |
||
| 152 | 'image': 'http://www.qq.com' |
||
| 153 | }] |
||
| 154 | result = self.client.message.send_articles(1, articles) |
||
| 155 | self.assertEqual(0, result['errcode']) |
||
| 156 | |||
| 157 | def test_send_card_message(self): |
||
| 158 | with HTTMock(wechat_api_mock): |
||
| 159 | result = self.client.message.send_card(1, '123456') |
||
| 160 | self.assertEqual(0, result['errcode']) |
||
| 161 | |||
| 162 | def test_send_mini_program_page(self): |
||
| 163 | with HTTMock(wechat_api_mock): |
||
| 164 | result = self.client.message.send_mini_program_page(1, {}) |
||
| 165 | self.assertEqual(0, result['errcode']) |
||
| 166 | |||
| 167 | def test_send_mass_text_message(self): |
||
| 168 | with HTTMock(wechat_api_mock): |
||
| 169 | result = self.client.message.send_mass_text([1], 'test', is_to_all=True) |
||
| 170 | self.assertEqual(0, result['errcode']) |
||
| 171 | |||
| 172 | def test_send_mass_image_message(self): |
||
| 173 | with HTTMock(wechat_api_mock): |
||
| 174 | result = self.client.message.send_mass_image([1], '123456', is_to_all=True) |
||
| 175 | self.assertEqual(0, result['errcode']) |
||
| 176 | |||
| 177 | def test_send_mass_voice_message(self): |
||
| 178 | with HTTMock(wechat_api_mock): |
||
| 179 | result = self.client.message.send_mass_voice([1], 'test', is_to_all=True) |
||
| 180 | self.assertEqual(0, result['errcode']) |
||
| 181 | |||
| 182 | def test_send_mass_video_message(self): |
||
| 183 | with HTTMock(wechat_api_mock): |
||
| 184 | result = self.client.message.send_mass_video([1], 'test', title='title', description='desc', is_to_all=True) |
||
| 185 | self.assertEqual(0, result['errcode']) |
||
| 186 | |||
| 187 | def test_send_mass_article_message(self): |
||
| 188 | with HTTMock(wechat_api_mock): |
||
| 189 | result = self.client.message.send_mass_article([1], 'test', is_to_all=True) |
||
| 190 | self.assertEqual(0, result['errcode']) |
||
| 191 | |||
| 192 | def test_send_mass_card_message(self): |
||
| 193 | with HTTMock(wechat_api_mock): |
||
| 194 | result = self.client.message.send_mass_card([1], 'test', is_to_all=True) |
||
| 195 | self.assertEqual(0, result['errcode']) |
||
| 196 | |||
| 197 | def test_get_mass_message(self): |
||
| 198 | with HTTMock(wechat_api_mock): |
||
| 199 | result = self.client.message.get_mass(201053012) |
||
| 200 | self.assertEqual('SEND_SUCCESS', result['msg_status']) |
||
| 201 | |||
| 202 | def test_create_menu(self): |
||
| 203 | with HTTMock(wechat_api_mock): |
||
| 204 | result = self.client.menu.create({ |
||
| 205 | 'button': [ |
||
| 206 | { |
||
| 207 | 'type': 'click', |
||
| 208 | 'name': 'test', |
||
| 209 | 'key': 'test' |
||
| 210 | } |
||
| 211 | ] |
||
| 212 | }) |
||
| 213 | self.assertEqual(0, result['errcode']) |
||
| 214 | |||
| 215 | def test_get_menu(self): |
||
| 216 | with HTTMock(wechat_api_mock): |
||
| 217 | menu = self.client.menu.get() |
||
| 218 | self.assertTrue('menu' in menu) |
||
| 219 | |||
| 220 | def test_delete_menu(self): |
||
| 221 | with HTTMock(wechat_api_mock): |
||
| 222 | result = self.client.menu.delete() |
||
| 223 | self.assertEqual(0, result['errcode']) |
||
| 224 | |||
| 225 | def test_update_menu(self): |
||
| 226 | with HTTMock(wechat_api_mock): |
||
| 227 | result = self.client.menu.update({ |
||
| 228 | 'button': [ |
||
| 229 | { |
||
| 230 | 'type': 'click', |
||
| 231 | 'name': 'test', |
||
| 232 | 'key': 'test' |
||
| 233 | } |
||
| 234 | ] |
||
| 235 | }) |
||
| 236 | self.assertEqual(0, result['errcode']) |
||
| 237 | |||
| 238 | def test_short_url(self): |
||
| 239 | with HTTMock(wechat_api_mock): |
||
| 240 | result = self.client.misc.short_url('http://www.qq.com') |
||
| 241 | self.assertEqual('http://qq.com', result['short_url']) |
||
| 242 | |||
| 243 | def test_get_wechat_ips(self): |
||
| 244 | with HTTMock(wechat_api_mock): |
||
| 245 | result = self.client.misc.get_wechat_ips() |
||
| 246 | self.assertEqual(['127.0.0.1'], result) |
||
| 247 | |||
| 248 | def test_get_user_info(self): |
||
| 249 | with HTTMock(wechat_api_mock): |
||
| 250 | openid = 'o6_bmjrPTlm6_2sgVt7hMZOPfL2M' |
||
| 251 | user = self.client.user.get(openid) |
||
| 252 | self.assertEqual('Band', user['nickname']) |
||
| 253 | |||
| 254 | def test_get_followers(self): |
||
| 255 | with HTTMock(wechat_api_mock): |
||
| 256 | result = self.client.user.get_followers() |
||
| 257 | self.assertEqual(2, result['total']) |
||
| 258 | self.assertEqual(2, result['count']) |
||
| 259 | |||
| 260 | def test_iter_followers(self): |
||
| 261 | @urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$', query=r'.*next_openid=[^&]+') |
||
| 262 | def next_openid_mock(url, request): |
||
| 263 | """伪造第二页的请求""" |
||
| 264 | content = { |
||
| 265 | "total": 2, |
||
| 266 | "count": 0, |
||
| 267 | "next_openid": "" |
||
| 268 | } |
||
| 269 | headers = { |
||
| 270 | 'Content-Type': 'application/json' |
||
| 271 | } |
||
| 272 | return response(200, content, headers, request=request) |
||
| 273 | |||
| 274 | with HTTMock(next_openid_mock, wechat_api_mock): |
||
| 275 | users = list(self.client.user.iter_followers()) |
||
| 276 | self.assertEqual(2, len(users)) |
||
| 277 | self.assertIn("OPENID1", users) |
||
| 278 | self.assertIn("OPENID2", users) |
||
| 279 | |||
| 280 | def test_update_user_remark(self): |
||
| 281 | with HTTMock(wechat_api_mock): |
||
| 282 | openid = 'openid' |
||
| 283 | remark = 'test' |
||
| 284 | result = self.client.user.update_remark(openid, remark) |
||
| 285 | self.assertEqual(0, result['errcode']) |
||
| 286 | |||
| 287 | def test_get_user_info_batch(self): |
||
| 288 | user_list = [ |
||
| 289 | { |
||
| 290 | "openid": "otvxTs4dckWG7imySrJd6jSi0CWE", |
||
| 291 | "lang": "zh-CN" |
||
| 292 | }, |
||
| 293 | { |
||
| 294 | "openid": "otvxTs_JZ6SEiP0imdhpi50fuSZg", |
||
| 295 | "lang": "zh-CN" |
||
| 296 | } |
||
| 297 | ] |
||
| 298 | with HTTMock(wechat_api_mock): |
||
| 299 | result = self.client.user.get_batch(user_list) |
||
| 300 | self.assertEqual(user_list[0]['openid'], result[0]['openid']) |
||
| 301 | self.assertEqual('iWithery', result[0]['nickname']) |
||
| 302 | self.assertEqual(user_list[1]['openid'], result[1]['openid']) |
||
| 303 | |||
| 304 | def test_get_user_info_batch_openid_list(self): |
||
| 305 | user_list = [ |
||
| 306 | 'otvxTs4dckWG7imySrJd6jSi0CWE', |
||
| 307 | 'otvxTs_JZ6SEiP0imdhpi50fuSZg' |
||
| 308 | ] |
||
| 309 | with HTTMock(wechat_api_mock): |
||
| 310 | result = self.client.user.get_batch(user_list) |
||
| 311 | self.assertEqual(user_list[0], result[0]['openid']) |
||
| 312 | self.assertEqual('iWithery', result[0]['nickname']) |
||
| 313 | self.assertEqual(user_list[1], result[1]['openid']) |
||
| 314 | |||
| 315 | def test_get_tag_users(self): |
||
| 316 | with HTTMock(wechat_api_mock): |
||
| 317 | result = self.client.tag.get_tag_users(101) |
||
| 318 | self.assertEqual(2, result['count']) |
||
| 319 | |||
| 320 | def test_iter_tag_users(self): |
||
| 321 | @urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$', path=r'.*user/tag/get') |
||
| 322 | def next_openid_mock(url, request): |
||
| 323 | """伪造第二页的请求""" |
||
| 324 | data = json.loads(request.body.decode()) |
||
| 325 | if not data.get('next_openid'): |
||
| 326 | return wechat_api_mock(url, request) |
||
| 327 | |||
| 328 | # 根据拿到的第二页请求响应 是没有data和next_openid的 |
||
| 329 | content = { |
||
| 330 | "count": 0 |
||
| 331 | } |
||
| 332 | headers = { |
||
| 333 | 'Content-Type': 'application/json' |
||
| 334 | } |
||
| 335 | return response(200, content, headers, request=request) |
||
| 336 | |||
| 337 | with HTTMock(next_openid_mock, wechat_api_mock): |
||
| 338 | users = list(self.client.tag.iter_tag_users(101)) |
||
| 339 | self.assertEqual(2, len(users)) |
||
| 340 | self.assertIn("OPENID1", users) |
||
| 341 | self.assertIn("OPENID2", users) |
||
| 342 | |||
| 343 | def test_create_qrcode(self): |
||
| 344 | data = { |
||
| 345 | 'expire_seconds': 1800, |
||
| 346 | 'action_name': 'QR_SCENE', |
||
| 347 | 'action_info': { |
||
| 348 | 'scene': {'scene_id': 123} |
||
| 349 | } |
||
| 350 | } |
||
| 351 | with HTTMock(wechat_api_mock): |
||
| 352 | result = self.client.qrcode.create(data) |
||
| 353 | self.assertEqual(1800, result['expire_seconds']) |
||
| 354 | |||
| 355 | def test_get_qrcode_url_with_str_ticket(self): |
||
| 356 | ticket = '123' |
||
| 357 | url = self.client.qrcode.get_url(ticket) |
||
| 358 | self.assertEqual( |
||
| 359 | 'https://mp.weixin.qq.com/cgi-bin/showqrcode?ticket=123', |
||
| 360 | url |
||
| 361 | ) |
||
| 362 | |||
| 363 | def test_get_qrcode_url_with_dict_ticket(self): |
||
| 364 | ticket = { |
||
| 365 | 'ticket': '123', |
||
| 366 | } |
||
| 367 | url = self.client.qrcode.get_url(ticket) |
||
| 368 | self.assertEqual( |
||
| 369 | 'https://mp.weixin.qq.com/cgi-bin/showqrcode?ticket=123', |
||
| 370 | url |
||
| 371 | ) |
||
| 372 | |||
| 373 | def test_customservice_add_account(self): |
||
| 374 | with HTTMock(wechat_api_mock): |
||
| 375 | result = self.client.customservice.add_account( |
||
| 376 | 'test1@test', |
||
| 377 | 'test1', |
||
| 378 | 'test1' |
||
| 379 | ) |
||
| 380 | self.assertEqual(0, result['errcode']) |
||
| 381 | |||
| 382 | def test_customservice_update_account(self): |
||
| 383 | with HTTMock(wechat_api_mock): |
||
| 384 | result = self.client.customservice.update_account( |
||
| 385 | 'test1@test', |
||
| 386 | 'test1', |
||
| 387 | 'test1' |
||
| 388 | ) |
||
| 389 | self.assertEqual(0, result['errcode']) |
||
| 390 | |||
| 391 | def test_customservice_delete_account(self): |
||
| 392 | with HTTMock(wechat_api_mock): |
||
| 393 | result = self.client.customservice.delete_account( |
||
| 394 | 'test1@test', |
||
| 395 | ) |
||
| 396 | self.assertEqual(0, result['errcode']) |
||
| 397 | |||
| 398 | def test_customservice_upload_headimg(self): |
||
| 399 | media_file = six.StringIO('nothing') |
||
| 400 | with HTTMock(wechat_api_mock): |
||
| 401 | result = self.client.customservice.upload_headimg( |
||
| 402 | 'test1@test', |
||
| 403 | media_file |
||
| 404 | ) |
||
| 405 | self.assertEqual(0, result['errcode']) |
||
| 406 | |||
| 407 | def test_customservice_get_accounts(self): |
||
| 408 | with HTTMock(wechat_api_mock): |
||
| 409 | result = self.client.customservice.get_accounts() |
||
| 410 | self.assertEqual(2, len(result)) |
||
| 411 | |||
| 412 | def test_customservice_get_online_accounts(self): |
||
| 413 | with HTTMock(wechat_api_mock): |
||
| 414 | result = self.client.customservice.get_online_accounts() |
||
| 415 | self.assertEqual(2, len(result)) |
||
| 416 | |||
| 417 | def test_customservice_create_session(self): |
||
| 418 | with HTTMock(wechat_api_mock): |
||
| 419 | result = self.client.customservice.create_session( |
||
| 420 | 'openid', |
||
| 421 | 'test1@test' |
||
| 422 | ) |
||
| 423 | self.assertEqual(0, result['errcode']) |
||
| 424 | |||
| 425 | def test_customservice_close_session(self): |
||
| 426 | with HTTMock(wechat_api_mock): |
||
| 427 | result = self.client.customservice.close_session( |
||
| 428 | 'openid', |
||
| 429 | 'test1@test' |
||
| 430 | ) |
||
| 431 | self.assertEqual(0, result['errcode']) |
||
| 432 | |||
| 433 | def test_customservice_get_session(self): |
||
| 434 | with HTTMock(wechat_api_mock): |
||
| 435 | result = self.client.customservice.get_session('openid') |
||
| 436 | self.assertEqual('test1@test', result['kf_account']) |
||
| 437 | |||
| 438 | def test_customservice_get_session_list(self): |
||
| 439 | with HTTMock(wechat_api_mock): |
||
| 440 | result = self.client.customservice.get_session_list('test1@test') |
||
| 441 | self.assertEqual(2, len(result)) |
||
| 442 | |||
| 443 | def test_customservice_get_wait_case(self): |
||
| 444 | with HTTMock(wechat_api_mock): |
||
| 445 | result = self.client.customservice.get_wait_case() |
||
| 446 | self.assertEqual(150, result['count']) |
||
| 447 | |||
| 448 | def test_customservice_get_records(self): |
||
| 449 | with HTTMock(wechat_api_mock): |
||
| 450 | result = self.client.customservice.get_records( |
||
| 451 | 123456789, |
||
| 452 | 987654321, |
||
| 453 | 1 |
||
| 454 | ) |
||
| 455 | self.assertEqual(2, len(result['recordlist'])) |
||
| 456 | |||
| 457 | def test_datacube_get_user_summary(self): |
||
| 458 | with HTTMock(wechat_api_mock): |
||
| 459 | result = self.client.datacube.get_user_summary( |
||
| 460 | '2014-12-06', |
||
| 461 | '2014-12-07' |
||
| 462 | ) |
||
| 463 | self.assertEqual(1, len(result)) |
||
| 464 | |||
| 465 | def test_datacube_get_user_cumulate(self): |
||
| 466 | with HTTMock(wechat_api_mock): |
||
| 467 | result = self.client.datacube.get_user_cumulate( |
||
| 468 | datetime(2014, 12, 6), |
||
| 469 | datetime(2014, 12, 7) |
||
| 470 | ) |
||
| 471 | self.assertEqual(1, len(result)) |
||
| 472 | |||
| 473 | def test_datacube_get_interface_summary(self): |
||
| 474 | with HTTMock(wechat_api_mock): |
||
| 475 | result = self.client.datacube.get_interface_summary( |
||
| 476 | '2014-12-06', |
||
| 477 | '2014-12-07' |
||
| 478 | ) |
||
| 479 | self.assertEqual(1, len(result)) |
||
| 480 | |||
| 481 | def test_datacube_get_interface_summary_hour(self): |
||
| 482 | with HTTMock(wechat_api_mock): |
||
| 483 | result = self.client.datacube.get_interface_summary_hour( |
||
| 484 | '2014-12-06', |
||
| 485 | '2014-12-07' |
||
| 486 | ) |
||
| 487 | self.assertEqual(1, len(result)) |
||
| 488 | |||
| 489 | def test_datacube_get_article_summary(self): |
||
| 490 | with HTTMock(wechat_api_mock): |
||
| 491 | result = self.client.datacube.get_article_summary( |
||
| 492 | '2014-12-06', |
||
| 493 | '2014-12-07' |
||
| 494 | ) |
||
| 495 | self.assertEqual(1, len(result)) |
||
| 496 | |||
| 497 | def test_datacube_get_article_total(self): |
||
| 498 | with HTTMock(wechat_api_mock): |
||
| 499 | result = self.client.datacube.get_article_total( |
||
| 500 | '2014-12-06', |
||
| 501 | '2014-12-07' |
||
| 502 | ) |
||
| 503 | self.assertEqual(1, len(result)) |
||
| 504 | |||
| 505 | def test_datacube_get_user_read(self): |
||
| 506 | with HTTMock(wechat_api_mock): |
||
| 507 | result = self.client.datacube.get_user_read( |
||
| 508 | '2014-12-06', |
||
| 509 | '2014-12-07' |
||
| 510 | ) |
||
| 511 | self.assertEqual(1, len(result)) |
||
| 512 | |||
| 513 | def test_datacube_get_user_read_hour(self): |
||
| 514 | with HTTMock(wechat_api_mock): |
||
| 515 | result = self.client.datacube.get_user_read_hour( |
||
| 516 | '2014-12-06', |
||
| 517 | '2014-12-07' |
||
| 518 | ) |
||
| 519 | self.assertEqual(1, len(result)) |
||
| 520 | |||
| 521 | def test_datacube_get_user_share(self): |
||
| 522 | with HTTMock(wechat_api_mock): |
||
| 523 | result = self.client.datacube.get_user_share( |
||
| 524 | '2014-12-06', |
||
| 525 | '2014-12-07' |
||
| 526 | ) |
||
| 527 | self.assertEqual(2, len(result)) |
||
| 528 | |||
| 529 | def test_datacube_get_user_share_hour(self): |
||
| 530 | with HTTMock(wechat_api_mock): |
||
| 531 | result = self.client.datacube.get_user_share_hour( |
||
| 532 | '2014-12-06', |
||
| 533 | '2014-12-07' |
||
| 534 | ) |
||
| 535 | self.assertEqual(1, len(result)) |
||
| 536 | |||
| 537 | def test_datacube_get_upstream_msg(self): |
||
| 538 | with HTTMock(wechat_api_mock): |
||
| 539 | result = self.client.datacube.get_upstream_msg( |
||
| 540 | '2014-12-06', |
||
| 541 | '2014-12-07' |
||
| 542 | ) |
||
| 543 | self.assertEqual(1, len(result)) |
||
| 544 | |||
| 545 | def test_datacube_get_upstream_msg_hour(self): |
||
| 546 | with HTTMock(wechat_api_mock): |
||
| 547 | result = self.client.datacube.get_upstream_msg_hour( |
||
| 548 | '2014-12-06', |
||
| 549 | '2014-12-07' |
||
| 550 | ) |
||
| 551 | self.assertEqual(1, len(result)) |
||
| 552 | |||
| 553 | def test_datacube_get_upstream_msg_week(self): |
||
| 554 | with HTTMock(wechat_api_mock): |
||
| 555 | result = self.client.datacube.get_upstream_msg_week( |
||
| 556 | '2014-12-06', |
||
| 557 | '2014-12-07' |
||
| 558 | ) |
||
| 559 | self.assertEqual(1, len(result)) |
||
| 560 | |||
| 561 | def test_datacube_get_upstream_msg_month(self): |
||
| 562 | with HTTMock(wechat_api_mock): |
||
| 563 | result = self.client.datacube.get_upstream_msg_month( |
||
| 564 | '2014-12-06', |
||
| 565 | '2014-12-07' |
||
| 566 | ) |
||
| 567 | self.assertEqual(1, len(result)) |
||
| 568 | |||
| 569 | def test_datacube_get_upstream_msg_dist(self): |
||
| 570 | with HTTMock(wechat_api_mock): |
||
| 571 | result = self.client.datacube.get_upstream_msg_dist( |
||
| 572 | '2014-12-06', |
||
| 573 | '2014-12-07' |
||
| 574 | ) |
||
| 575 | self.assertEqual(1, len(result)) |
||
| 576 | |||
| 577 | def test_datacube_get_upstream_msg_dist_week(self): |
||
| 578 | with HTTMock(wechat_api_mock): |
||
| 579 | result = self.client.datacube.get_upstream_msg_dist_week( |
||
| 580 | '2014-12-06', |
||
| 581 | '2014-12-07' |
||
| 582 | ) |
||
| 583 | self.assertEqual(1, len(result)) |
||
| 584 | |||
| 585 | def test_datacube_get_upstream_msg_dist_month(self): |
||
| 586 | with HTTMock(wechat_api_mock): |
||
| 587 | result = self.client.datacube.get_upstream_msg_dist_month( |
||
| 588 | '2014-12-06', |
||
| 589 | '2014-12-07' |
||
| 590 | ) |
||
| 591 | self.assertEqual(1, len(result)) |
||
| 592 | |||
| 593 | def test_jsapi_get_ticket_response(self): |
||
| 594 | with HTTMock(wechat_api_mock): |
||
| 595 | result = self.client.jsapi.get_ticket() |
||
| 596 | self.assertEqual( |
||
| 597 | 'bxLdikRXVbTPdHSM05e5u5sUoXNKd8-41ZO3MhKoyN5OfkWITDGgnr2fwJ0m9E8NYzWKVZvdVtaUgWvsdshFKA', # NOQA |
||
| 598 | result['ticket'] |
||
| 599 | ) |
||
| 600 | self.assertEqual(7200, result['expires_in']) |
||
| 601 | |||
| 602 | def test_jsapi_get_jsapi_signature(self): |
||
| 603 | noncestr = 'Wm3WZYTPz0wzccnW' |
||
| 604 | ticket = 'sM4AOVdWfPE4DxkXGEs8VMCPGGVi4C3VM0P37wVUCFvkVAy_90u5h9nbSlYy3-Sl-HhTdfl2fzFy1AOcHKP7qg' # NOQA |
||
| 605 | timestamp = 1414587457 |
||
| 606 | url = 'http://mp.weixin.qq.com?params=value' |
||
| 607 | signature = self.client.jsapi.get_jsapi_signature( |
||
| 608 | noncestr, |
||
| 609 | ticket, |
||
| 610 | timestamp, |
||
| 611 | url |
||
| 612 | ) |
||
| 613 | self.assertEqual( |
||
| 614 | '0f9de62fce790f9a083d5c99e95740ceb90c27ed', |
||
| 615 | signature |
||
| 616 | ) |
||
| 617 | |||
| 618 | def test_jsapi_get_jsapi_card_ticket(self): |
||
| 619 | """card_ticket 与 jsapi_ticket 的 api 都相同，除了请求参数 type 为 wx_card |
||
| 620 | 所以这里使用与 `test_jsapi_get_ticket` 相同的测试文件""" |
||
| 621 | with HTTMock(wechat_api_mock): |
||
| 622 | ticket = self.client.jsapi.get_jsapi_card_ticket() |
||
| 623 | self.assertEqual( |
||
| 624 | 'bxLdikRXVbTPdHSM05e5u5sUoXNKd8-41ZO3MhKoyN5OfkWITDGgnr2fwJ0m9E8NYzWKVZvdVtaUgWvsdshFKA', # NOQA |
||
| 625 | ticket |
||
| 626 | ) |
||
| 627 | self.assertTrue( |
||
| 628 | 7200 < self.client.session.get('{0}_jsapi_card_ticket_expires_at'.format(self.client.appid)) |
||
| 629 | ) |
||
| 630 | self.assertEqual( |
||
| 631 | self.client.session.get('{0}_jsapi_card_ticket'.format(self.client.appid)), |
||
| 632 | 'bxLdikRXVbTPdHSM05e5u5sUoXNKd8-41ZO3MhKoyN5OfkWITDGgnr2fwJ0m9E8NYzWKVZvdVtaUgWvsdshFKA', |
||
| 633 | ) |
||
| 634 | |||
| 635 | def test_jsapi_get_jsapi_card_params(self): |
||
| 636 | """微信签名测试工具：http://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=cardsign""" |
||
| 637 | noncestr = 'Wm3WZYTPz0wzccnW' |
||
| 638 | card_ticket = 'sM4AOVdWfPE4DxkXGEs8VMCPGGVi4C3VM0P37wVUCFvkVAy_90u5h9nbSlYy3-Sl-HhTdfl2fzFy1AOcHKP7qg' |
||
| 639 | timestamp = 1414587457 |
||
| 640 | signature_dict = self.client.jsapi.get_jsapi_card_params( |
||
| 641 | noncestr=noncestr, |
||
| 642 | card_ticket=card_ticket, |
||
| 643 | timestamp=timestamp, |
||
| 644 | card_type='GROUPON', |
||
| 645 | ) |
||
| 646 | self.assertEqual( |
||
| 647 | {'card_type': 'GROUPON', 'noncestr': 'Wm3WZYTPz0wzccnW', |
||
| 648 | 'api_ticket': 'sM4AOVdWfPE4DxkXGEs8VMCPGGVi4C3VM0P37wVUCFvkVAy_90u5h9nbSlYy3-Sl-HhTdfl2fzFy1AOcHKP7qg', |
||
| 649 | 'appid': '123456', 'timestamp': 1414587457, 'sign': 'c47b1fb500eb35d8f2f9b9375b4491089df953e2'}, |
||
| 650 | signature_dict |
||
| 651 | ) |
||
| 652 | |||
| 653 | def test_menu_get_menu_info(self): |
||
| 654 | with HTTMock(wechat_api_mock): |
||
| 655 | menu_info = self.client.menu.get_menu_info() |
||
| 656 | self.assertEqual(1, menu_info['is_menu_open']) |
||
| 657 | |||
| 658 | def test_message_get_autoreply_info(self): |
||
| 659 | with HTTMock(wechat_api_mock): |
||
| 660 | autoreply = self.client.message.get_autoreply_info() |
||
| 661 | self.assertEqual(1, autoreply['is_autoreply_open']) |
||
| 662 | |||
| 663 | def test_shakearound_apply_device_id(self): |
||
| 664 | with HTTMock(wechat_api_mock): |
||
| 665 | res = self.client.shakearound.apply_device_id(1, 'test') |
||
| 666 | self.assertEqual(123, res['apply_id']) |
||
| 667 | |||
| 668 | def test_shakearound_update_device(self): |
||
| 669 | with HTTMock(wechat_api_mock): |
||
| 670 | res = self.client.shakearound.update_device('1234', comment='test') |
||
| 671 | self.assertEqual(0, res['errcode']) |
||
| 672 | |||
| 673 | def test_shakearound_bind_device_location(self): |
||
| 674 | with HTTMock(wechat_api_mock): |
||
| 675 | res = self.client.shakearound.bind_device_location(123, 1234) |
||
| 676 | self.assertEqual(0, res['errcode']) |
||
| 677 | |||
| 678 | def test_shakearound_search_device(self): |
||
| 679 | with HTTMock(wechat_api_mock): |
||
| 680 | res = self.client.shakearound.search_device(apply_id=123) |
||
| 681 | self.assertEqual(151, res['total_count']) |
||
| 682 | self.assertEqual(2, len(res['devices'])) |
||
| 683 | |||
| 684 | def test_shakearound_add_page(self): |
||
| 685 | with HTTMock(wechat_api_mock): |
||
| 686 | res = self.client.shakearound.add_page( |
||
| 687 | 'test', |
||
| 688 | 'test', |
||
| 689 | 'http://www.qq.com', |
||
| 690 | 'http://www.qq.com' |
||
| 691 | ) |
||
| 692 | self.assertEqual(28840, res['page_id']) |
||
| 693 | |||
| 694 | def test_shakearound_update_page(self): |
||
| 695 | with HTTMock(wechat_api_mock): |
||
| 696 | res = self.client.shakearound.update_page( |
||
| 697 | 123, |
||
| 698 | 'test', |
||
| 699 | 'test', |
||
| 700 | 'http://www.qq.com', |
||
| 701 | 'http://www.qq.com' |
||
| 702 | ) |
||
| 703 | self.assertEqual(28840, res['page_id']) |
||
| 704 | |||
| 705 | def test_shakearound_delete_page(self): |
||
| 706 | with HTTMock(wechat_api_mock): |
||
| 707 | res = self.client.shakearound.delete_page(123) |
||
| 708 | self.assertEqual(0, res['errcode']) |
||
| 709 | |||
| 710 | def test_shakearound_search_page(self): |
||
| 711 | with HTTMock(wechat_api_mock): |
||
| 712 | res = self.client.shakearound.search_pages(123) |
||
| 713 | self.assertEqual(2, res['total_count']) |
||
| 714 | self.assertEqual(2, len(res['pages'])) |
||
| 715 | |||
| 716 | def test_shakearound_add_material(self): |
||
| 717 | with HTTMock(wechat_api_mock): |
||
| 718 | media_file = six.StringIO('nothing') |
||
| 719 | res = self.client.shakearound.add_material(media_file, 'icon') |
||
| 720 | self.assertEqual( |
||
| 721 | 'http://shp.qpic.cn/wechat_shakearound_pic/0/1428377032e9dd2797018cad79186e03e8c5aec8dc/120', # NOQA |
||
| 722 | res['pic_url'] |
||
| 723 | ) |
||
| 724 | |||
| 725 | def test_shakearound_bind_device_pages(self): |
||
| 726 | with HTTMock(wechat_api_mock): |
||
| 727 | res = self.client.shakearound.bind_device_pages( |
||
| 728 | 123, |
||
| 729 | 1, |
||
| 730 | 1, |
||
| 731 | 1234 |
||
| 732 | ) |
||
| 733 | self.assertEqual(0, res['errcode']) |
||
| 734 | |||
| 735 | def test_shakearound_get_shake_info(self): |
||
| 736 | with HTTMock(wechat_api_mock): |
||
| 737 | res = self.client.shakearound.get_shake_info('123456') |
||
| 738 | self.assertEqual(14211, res['page_id']) |
||
| 739 | self.assertEqual('oVDmXjp7y8aG2AlBuRpMZTb1-cmA', res['openid']) |
||
| 740 | |||
| 741 | def test_shakearound_get_device_statistics(self): |
||
| 742 | with HTTMock(wechat_api_mock): |
||
| 743 | res = self.client.shakearound.get_device_statistics( |
||
| 744 | '2015-04-01 00:00:00', |
||
| 745 | '2015-04-17 00:00:00', |
||
| 746 | 1234 |
||
| 747 | ) |
||
| 748 | self.assertEqual(2, len(res)) |
||
| 749 | |||
| 750 | def test_shakearound_get_page_statistics(self): |
||
| 751 | with HTTMock(wechat_api_mock): |
||
| 752 | res = self.client.shakearound.get_page_statistics( |
||
| 753 | '2015-04-01 00:00:00', |
||
| 754 | '2015-04-17 00:00:00', |
||
| 755 | 1234 |
||
| 756 | ) |
||
| 757 | self.assertEqual(2, len(res)) |
||
| 758 | |||
| 759 | def test_material_get_count(self): |
||
| 760 | with HTTMock(wechat_api_mock): |
||
| 761 | res = self.client.material.get_count() |
||
| 762 | self.assertEqual(1, res['voice_count']) |
||
| 763 | self.assertEqual(2, res['video_count']) |
||
| 764 | self.assertEqual(3, res['image_count']) |
||
| 765 | self.assertEqual(4, res['news_count']) |
||
| 766 | |||
| 767 | def test_shakearound_get_apply_status(self): |
||
| 768 | with HTTMock(wechat_api_mock): |
||
| 769 | res = self.client.shakearound.get_apply_status( |
||
| 770 | 1234 |
||
| 771 | ) |
||
| 772 | self.assertEqual(4, len(res)) |
||
| 773 | |||
| 774 | View Code Duplication | def test_reraise_requests_exception(self): |
|
|
Duplication
introduced
by
|
|||
| 775 | @urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$') |
||
| 776 | def _wechat_api_mock(url, request): |
||
| 777 | return {'status_code': 404, 'content': '404 not found'} |
||
| 778 | |||
| 779 | try: |
||
| 780 | with HTTMock(_wechat_api_mock): |
||
| 781 | self.client.material.get_count() |
||
| 782 | except WeChatClientException as e: |
||
| 783 | self.assertEqual(404, e.response.status_code) |
||
| 784 | |||
| 785 | def test_wifi_list_shops(self): |
||
| 786 | with HTTMock(wechat_api_mock): |
||
| 787 | res = self.client.wifi.list_shops() |
||
| 788 | self.assertEqual(16, res['totalcount']) |
||
| 789 | self.assertEqual(1, res['pageindex']) |
||
| 790 | |||
| 791 | def test_wifi_get_shop(self): |
||
| 792 | with HTTMock(wechat_api_mock): |
||
| 793 | res = self.client.wifi.get_shop(1) |
||
| 794 | self.assertEqual(1, res['bar_type']) |
||
| 795 | self.assertEqual(2, res['ap_count']) |
||
| 796 | |||
| 797 | def test_wifi_add_device(self): |
||
| 798 | with HTTMock(wechat_api_mock): |
||
| 799 | res = self.client.wifi.add_device( |
||
| 800 | 123, 'WX-test', '12345678', '00:1f:7a:ad:5c:a8' |
||
| 801 | ) |
||
| 802 | self.assertEqual(0, res['errcode']) |
||
| 803 | |||
| 804 | def test_wifi_list_devices(self): |
||
| 805 | with HTTMock(wechat_api_mock): |
||
| 806 | res = self.client.wifi.list_devices() |
||
| 807 | self.assertEqual(2, res['totalcount']) |
||
| 808 | self.assertEqual(1, res['pageindex']) |
||
| 809 | |||
| 810 | def test_wifi_delete_device(self): |
||
| 811 | with HTTMock(wechat_api_mock): |
||
| 812 | res = self.client.wifi.delete_device('00:1f:7a:ad:5c:a8') |
||
| 813 | self.assertEqual(0, res['errcode']) |
||
| 814 | |||
| 815 | def test_wifi_get_qrcode_url(self): |
||
| 816 | with HTTMock(wechat_api_mock): |
||
| 817 | qrcode_url = self.client.wifi.get_qrcode_url(123, 0) |
||
| 818 | self.assertEqual('http://www.qq.com', qrcode_url) |
||
| 819 | |||
| 820 | def test_wifi_set_homepage(self): |
||
| 821 | with HTTMock(wechat_api_mock): |
||
| 822 | res = self.client.wifi.set_homepage(123, 0) |
||
| 823 | self.assertEqual(0, res['errcode']) |
||
| 824 | |||
| 825 | def test_wifi_get_homepage(self): |
||
| 826 | with HTTMock(wechat_api_mock): |
||
| 827 | res = self.client.wifi.get_homepage(429620) |
||
| 828 | self.assertEqual(1, res['template_id']) |
||
| 829 | self.assertEqual('http://wifi.weixin.qq.com/', res['url']) |
||
| 830 | |||
| 831 | def test_wifi_list_statistics(self): |
||
| 832 | with HTTMock(wechat_api_mock): |
||
| 833 | res = self.client.wifi.list_statistics('2015-05-01', '2015-05-02') |
||
| 834 | self.assertEqual(2, len(res)) |
||
| 835 | |||
| 836 | def test_upload_mass_image(self): |
||
| 837 | media_file = six.StringIO('nothing') |
||
| 838 | with HTTMock(wechat_api_mock): |
||
| 839 | res = self.client.media.upload_mass_image(media_file) |
||
| 840 | self.assertEqual( |
||
| 841 | 'http://mmbiz.qpic.cn/mmbiz/gLO17UPS6FS2xsypf378iaNhWacZ1G1UplZYWEYfwvuU6Ont96b1roYs CNFwaRrSaKTPCUdBK9DgEHicsKwWCBRQ/0', # NOQA |
||
| 842 | res |
||
| 843 | ) |
||
| 844 | |||
| 845 | def test_scan_get_merchant_info(self): |
||
| 846 | with HTTMock(wechat_api_mock): |
||
| 847 | res = self.client.scan.get_merchant_info() |
||
| 848 | self.assertEqual(8888, res['verified_firm_code_list'][0]) |
||
| 849 | |||
| 850 | def test_scan_create_product(self): |
||
| 851 | with HTTMock(wechat_api_mock): |
||
| 852 | res = self.client.scan.create_product({ |
||
| 853 | "keystandard": "ean13", |
||
| 854 | "keystr": "6900000000000", |
||
| 855 | }) |
||
| 856 | self.assertEqual('5g0B4A90aqc', res['pid']) |
||
| 857 | |||
| 858 | def test_scan_publish_product(self): |
||
| 859 | with HTTMock(wechat_api_mock): |
||
| 860 | res = self.client.scan.publish_product('ean13', '6900873042720') |
||
| 861 | self.assertEqual(0, res['errcode']) |
||
| 862 | |||
| 863 | def test_scan_unpublish_product(self): |
||
| 864 | with HTTMock(wechat_api_mock): |
||
| 865 | res = self.client.scan.unpublish_product('ean13', '6900873042720') |
||
| 866 | self.assertEqual(0, res['errcode']) |
||
| 867 | |||
| 868 | def test_scan_set_test_whitelist(self): |
||
| 869 | with HTTMock(wechat_api_mock): |
||
| 870 | res = self.client.scan.set_test_whitelist(['openid1'], ['messense']) |
||
| 871 | self.assertEqual(0, res['errcode']) |
||
| 872 | |||
| 873 | def test_scan_get_product(self): |
||
| 874 | with HTTMock(wechat_api_mock): |
||
| 875 | res = self.client.scan.get_product('ean13', '6900873042720') |
||
| 876 | assert 'brand_info' in res |
||
| 877 | |||
| 878 | def test_scan_list_product(self): |
||
| 879 | with HTTMock(wechat_api_mock): |
||
| 880 | res = self.client.scan.list_product() |
||
| 881 | self.assertEqual(2, res['total']) |
||
| 882 | |||
| 883 | def test_scan_update_product(self): |
||
| 884 | with HTTMock(wechat_api_mock): |
||
| 885 | res = self.client.scan.update_product({ |
||
| 886 | "keystandard": "ean13", |
||
| 887 | "keystr": "6900000000000", |
||
| 888 | }) |
||
| 889 | self.assertEqual('5g0B4A90aqc', res['pid']) |
||
| 890 | |||
| 891 | def test_scan_clear_product(self): |
||
| 892 | with HTTMock(wechat_api_mock): |
||
| 893 | res = self.client.scan.clear_product('ean13', '6900873042720') |
||
| 894 | self.assertEqual(0, res['errcode']) |
||
| 895 | |||
| 896 | def test_scan_check_ticket(self): |
||
| 897 | with HTTMock(wechat_api_mock): |
||
| 898 | res = self.client.scan.check_ticket('Ym1haDlvNXJqY3Ru1') |
||
| 899 | self.assertEqual('otAzGjrS4AYCmeJM1GhEOcHXXTAo', res['openid']) |
||
| 900 | |||
| 901 | def test_change_openid(self): |
||
| 902 | with HTTMock(wechat_api_mock): |
||
| 903 | res = self.client.user.change_openid( |
||
| 904 | 'xxxxx', ['oEmYbwN-n24jxvk4Sox81qedINkQ', 'oEmYbwH9uVd4RKJk7ZZg6SzL6tTo'] |
||
| 905 | ) |
||
| 906 | self.assertEqual(2, len(res)) |
||
| 907 | self.assertEqual('o2FwqwI9xCsVadFah_HtpPfaR-X4', res[0]['new_openid']) |
||
| 908 | self.assertEqual('ori_openid error', res[1]['err_msg']) |
||
| 909 | |||
| 910 | def test_code_to_session(self): |
||
| 911 | with HTTMock(wechat_api_mock): |
||
| 912 | res = self.client.wxa.code_to_session('023dUeGW1oeGOZ0JXvHW1SDVFW1dUeGu') |
||
| 913 | self.assertIn('session_key', res) |
||
| 914 | self.assertEqual('D1ZWEygStjuLCnZ9IN2l4Q==', res['session_key']) |
||
| 915 | self.assertEqual('o16wA0b4AZKzgVJR3MBwoUdTfU_E', res['openid']) |
||
| 916 | self.assertEqual('or4zX05h_Ykt4ju0TUfx3CQsvfTo', res['unionid']) |
||
| 917 |
Loading history...