اختبار الحمل بالجراد. الجزء 2

بالنسبة لأولئك الذين أحبوا مقالتي السابقة ، ما زلت أشارك انطباعاتي عن أداة اختبار إجهاد الجراد.

سأحاول أن أوضح بوضوح مزايا كتابة ثعبان اختبار الحمل مع رمز يمكنك فيه بسهولة إعداد أي بيانات للاختبار ومعالجة النتائج.

معالجة استجابة الخادم

في بعض الأحيان في اختبار التحميل ، لا يكفي مجرد الحصول على 200 OK من خادم HTTP. يحدث ، فمن الضروري التحقق من محتويات الاستجابة للتأكد من أنه تحت التحميل يقوم الخادم بإصدار البيانات الصحيحة أو إجراء حسابات دقيقة. لمثل هذه الحالات فقط ، أضاف Locust القدرة على تجاوز معلمات نجاح استجابة الخادم. خذ بعين الاعتبار المثال التالي:

from locust import HttpLocust, TaskSet, task import random as rnd class UserBehavior(TaskSet): @task(1) def check_albums(self): photo_id = rnd.randint(1, 5000) with self.client.get(f'/photos/{photo_id}', catch_response=True, name='/photos/[id]') as response: if response.status_code == 200: album_id = response.json().get('albumId') if album_id % 10 != 0: response.success() else: response.failure(f'album id cannot be {album_id}') else: response.failure(f'status code is {response.status_code}') class WebsiteUser(HttpLocust): task_set = UserBehavior min_wait = 1000 max_wait = 2000 

لديها طلب واحد فقط ، مما سيؤدي إلى إنشاء تحميل في السيناريو التالي:

من الخادم ، نطلب كائنات صور ذات هوية عشوائية في النطاق من 1 إلى 5000 ونتحقق من معرف الألبوم في هذا الكائن ، بافتراض أنه لا يمكن أن يكون مضاعفًا لـ 10

هنا يمكنك إعطاء بعض التفسيرات على الفور:

• البناء الرائع مع الطلب () كرد: يمكنك استبداله بنجاح برد = request () والعمل بهدوء مع كائن الاستجابة
• يتم تكوين عنوان URL باستخدام صيغة تنسيق السلسلة المضافة في python 3.6 ، إذا لم أكن مخطئًا - f '/ photos / {photo_id}' . في الإصدارات السابقة ، لن يعمل هذا التصميم!
• حجة جديدة لم نستخدمها من قبل ، catch_response = True ، تخبر Locust أننا أنفسنا سنحدد نجاح استجابة الخادم. إذا لم تحدده ، فسوف نتلقى كائن الاستجابة بنفس الطريقة وسنكون قادرين على معالجة بياناته ، ولكن لن نعيد تعريف النتيجة. فيما يلي مثال مفصل.
• اسم وسيطة آخر = '/ photos / [id]' . هناك حاجة إلى تجميع الطلبات في الإحصائيات. يمكن أن يكون الاسم أي نص ، وليس من الضروري تكرار عنوان URL. بدونه ، سيتم تسجيل كل طلب بعنوان أو معلمات فريدة بشكل منفصل. إليك كيفية العمل:
  

باستخدام الحجة نفسها ، يمكنك تنفيذ خدعة أخرى - في بعض الأحيان يحدث أن تؤدي خدمة واحدة ذات معلمات مختلفة (على سبيل المثال ، محتويات مختلفة لطلبات POST) إلى منطق مختلف. حتى لا تختلط نتائج الاختبار ، يمكنك كتابة عدة مهام منفصلة ، مع تحديد اسم الوسيط الخاص بكل منها.

بعد ذلك نقوم بعمل الشيكات. لدي 2. أولا ، نتحقق من أن الخادم أعاد إجابة إذا response.status_code == 200 :

إذا كانت الإجابة بنعم ، فتحقق مما إذا كان معرف الألبوم مضاعفًا لعدد 10. وإذا لم يكن مضاعفًا ، فضع علامة على هذه الإجابة على أنها استجابة ناجحة.

في حالات أخرى ، نشير إلى سبب فشل الاستجابة. فشل ("نص الخطأ") . سيتم عرض هذا النص في صفحة الإخفاقات أثناء الاختبار.



أيضا ، يمكن للقراء اليقظين ملاحظة عدم وجود استثناءات مميزة للكود الذي يعمل مع واجهات الشبكة. في الواقع ، في حالة انتهاء المهلة وخطأ الاتصال وحوادث أخرى غير متوقعة ، سيتعامل الجراد مع الأخطاء وسيستمر في رد استجابة ، مشيرًا إلى أن حالة رمز الاستجابة هي 0.

إذا استمر الرمز في طرح استثناء ، فسيتم كتابته إلى علامة التبويب استثناءات في وقت التشغيل حتى نتمكن من التعامل معه. الوضع الأكثر شيوعًا هو أن json'e للإجابة لم تعيد القيمة التي كنا نبحث عنها ، لكننا نجري بالفعل العمليات التالية عليها.

قبل إغلاق الموضوع - في المثال أستخدم خادم json للتوضيح ، لأنه من الأسهل معالجة الاستجابات. ولكن يمكنك العمل بنفس النجاح مع HTML و XML و FormData ومرفقات الملفات والبيانات الأخرى التي تستخدمها البروتوكولات المستندة إلى HTTP.

العمل مع السيناريوهات المعقدة

في كل مرة تقريبًا تكون المهمة هي إجراء اختبار تحميل لتطبيق ويب ، يصبح من الواضح بسرعة أنه من المستحيل توفير تغطية لائقة مع خدمات GET وحدها - والتي ببساطة تعيد البيانات.

مثال كلاسيكي: لاختبار متجر على الإنترنت ، من المستحسن أن المستخدم

1. فتح المتجر الرئيسي
2. كنت أبحث عن البضائع
3. تفاصيل العنصر المفتوح
4. تمت إضافة عنصر إلى سلة التسوق
5. مدفوع

من المثال ، يمكننا أن نفترض أن خدمات الاتصال بترتيب عشوائي لن تعمل ، بالتسلسل فقط. علاوة على ذلك ، قد تحتوي السلع والسلال وطرق الدفع على معرفات فريدة لكل مستخدم.

باستخدام المثال السابق ، مع تعديلات طفيفة ، يمكنك بسهولة تنفيذ اختبار مثل هذا السيناريو. نقوم بتكييف المثال لخادم الاختبار لدينا:

1. يكتب المستخدم مشاركة جديدة
2. يكتب المستخدم تعليقًا على منشور جديد
3. يقرأ المستخدم التعليق

 from locust import HttpLocust, TaskSet, task class FlowException(Exception): pass class UserBehavior(TaskSet): @task(1) def check_flow(self): # step 1 new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'} post_response = self.client.post('/posts', json=new_post) if post_response.status_code != 201: raise FlowException('post not created') post_id = post_response.json().get('id') # step 2 new_comment = { "postId": post_id, "name": "my comment", "email": "test@user.habr", "body": "Author is cool. Some text. Hello world!" } comment_response = self.client.post('/comments', json=new_comment) if comment_response.status_code != 201: raise FlowException('comment not created') comment_id = comment_response.json().get('id') # step 3 self.client.get(f'/comments/{comment_id}', name='/comments/[id]') if comment_response.status_code != 200: raise FlowException('comment not read') class WebsiteUser(HttpLocust): task_set = UserBehavior min_wait = 1000 max_wait = 2000 

في هذا المثال ، أضفت فئة FlowException جديدة. بعد كل خطوة ، إذا لم تسر كما هو متوقع ، فأنا أرمي فئة الاستثناء هذه لمقاطعة البرنامج النصي - إذا لم تنجح المشاركة ، فلن يكون هناك أي تعليق ، إلخ. إذا رغبت في ذلك ، يمكن استبدال البناء بالعودة المعتادة ، ولكن في هذه الحالة ، في وقت التشغيل وعند تحليل النتائج ، لن يكون واضحًا جدًا حيث يقع البرنامج النصي المنفذة في علامة التبويب استثناءات. لنفس السبب ، لا أستخدم المحاولة ... باستثناء الإنشاء.

جعل الحمل واقعيًا

الآن يمكنني أن أتألم - في حالة المتجر كل شيء خطي حقًا ، لكن المثال مع المنشورات والتعليقات بعيد المنال - فهم يقرؤون المنشورات أكثر من 10 مرات أكثر مما يقومون بإنشائه. بشكل معقول ، دعنا نجعل المثال أكثر قابلية للتطبيق. وهناك نهجان على الأقل:

1. يمكنك "ترميز" قائمة المنشورات التي يقرأها المستخدمون وتبسيط رمز الاختبار إذا كان هناك مثل هذا الاحتمال ولا تعتمد وظائف الواجهة الخلفية على مشاركات محددة
2. احفظ المنشورات التي تم إنشاؤها وقراءتها إذا لم يكن من الممكن تعيين قائمة المنشورات مسبقًا أو يعتمد الحمل الواقعي على المشاركات التي تتم قراءتها (قمت بإزالة إنشاء التعليقات من المثال لجعل رمزه أصغر وأكثر مرئية)

 from locust import HttpLocust, TaskSet, task import random as r class UserBehavior(TaskSet): created_posts = [] @task(1) def create_post(self): new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'} post_response = self.client.post('/posts', json=new_post) if post_response.status_code != 201: return post_id = post_response.json().get('id') self.created_posts.append(post_id) @task(10) def read_post(self): if len(self.created_posts) == 0: return post_id = r.choice(self.created_posts) self.client.get(f'/posts/{post_id}', name='read post') class WebsiteUser(HttpLocust): task_set = UserBehavior min_wait = 1000 max_wait = 2000 

في فئة UserBehavior ، قمت بإنشاء قائمة create_posts . انتبه جيدًا - هذا كائن ولم يتم إنشاؤه في مُنشئ فئة __init __ () ، وبالتالي ، على عكس جلسة العميل ، فإن هذه القائمة شائعة لجميع المستخدمين. تقوم المهمة الأولى بإنشاء منشور وكتابة معرفه إلى القائمة. الثانية - 10 مرات في كثير من الأحيان ، يقرأ منشورًا تم اختياره عشوائيًا من القائمة. الشرط الإضافي للمهمة الثانية هو التحقق مما إذا تم إنشاء أي مشاركات.

إذا أردنا أن يعمل كل مستخدم ببياناته الخاصة فقط ، فيمكننا إعلانها في المنشئ على النحو التالي:

 class UserBehavior(TaskSet): def __init__(self, parent): super(UserBehavior, self).__init__(parent) self.created_posts = list() 

المزيد من الميزات

للإطلاق المتسلسل للمهام ، تقترح الوثائق الرسمية أننا نستخدم أيضًا التعليق التوضيحي للمهمةseq_task (1) ، مع تحديد الرقم التسلسلي للمهمة في الوسيطة

 class MyTaskSequence(TaskSequence): @seq_task(1) def first_task(self): pass @seq_task(2) def second_task(self): pass @seq_task(3) @task(10) def third_task(self): pass 

في هذا المثال ، سيقوم كل مستخدم أولاً بتنفيذ first_task ، ثم second_task ، ثم 10 مرة third_task .

بصراحة ، توفر هذه الفرصة يرضي ، ولكن ، على عكس الأمثلة السابقة ، ليس من الواضح كيفية نقل نتائج المهمة الأولى إلى الثانية إذا لزم الأمر.

أيضًا ، بالنسبة للسيناريوهات المعقدة بشكل خاص ، من الممكن إنشاء مجموعات مهام متداخلة ، في الواقع ، إنشاء العديد من فئات TaskSet والاتصال ببعضها البعض.

 from locust import HttpLocust, TaskSet, task class Todo(TaskSet): @task(3) def index(self): self.client.get("/todos") @task(1) def stop(self): self.interrupt() class UserBehavior(TaskSet): tasks = {Todo: 1} @task(3) def index(self): self.client.get("/") @task(2) def posts(self): self.client.get("/posts") class WebsiteUser(HttpLocust): task_set = UserBehavior min_wait = 1000 max_wait = 2000 

في المثال أعلاه ، مع احتمال من 1 إلى 6 ، سيتم تشغيل البرنامج النصي Todo ، وسيتم تنفيذه حتى الاحتمال من 1 إلى 4 ، سيعود إلى البرنامج النصي UserBehavior . من المهم جدًا أن يكون لديك دعوة إلى self.interrupt () - بدونها ، سيركز الاختبار على المهمة الفرعية.

شكرا للقراءة. في المقالة النهائية سأكتب عن الاختبارات الموزعة والاختبار بدون واجهة مستخدم ، وكذلك عن الصعوبات التي واجهتها أثناء الاختبار مع الجراد وكيفية الالتفاف عليها.