Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use untag_user method in localstack

Best Python code snippet using localstack_python

test_model_usertag.py

Source:test_model_usertag.py

1# -*- coding: utf-8 -*-2from __future__ import unicode_literals3from django.db import transaction4from wechatpy.client.api import WeChatGroup, WeChatTag5from ..models import UserTag, WeChatUser6from .base import mock, WeChatTestCase7class UserTestCase(WeChatTestCase):8    def test_sync(self):9        """æµè¯åæ¥ç¨æ·æ ç¾"""10        def assertTagSyncSuccess(custom_tags):11            with mock.patch.object(WeChatGroup, "get"),\12                mock.patch.object(WeChatTag, "update"),\13                mock.patch.object(WeChatTag, "create"):14                remote_tags = self.base_tags[:] + [15                    dict(16                        id=id,17                        name=name,18                        count=019                    )20                    for id, name in custom_tags.items()21                ]22                WeChatGroup.get.return_value = remote_tags23                # ä¸ä¼è°ç¨è¿ç¨æ¥å£24                WeChatTag.update.side_effect = Exception()25                WeChatTag.create.side_effect = Exception()26                UserTag.sync(self.app)27                tags = UserTag.objects.filter(app=self.app).all()28                self.assertEqual(len(tags), len(remote_tags))29                # æ£æ¥æ¯ä¸ªtagè¢«æ£ç¡®æå¥30                for tag in remote_tags:31                    UserTag.objects.get(app=self.app, id=tag["id"], name=tag["name"])32        assertTagSyncSuccess({101: "group1", 102: "group2"})33        assertTagSyncSuccess({103: "group1", 102: "group2"})34        assertTagSyncSuccess({104: "group3"})35    def test_sync_users(self):36        """æµè¯åæ¥æ ç¾ä¸çç¨æ·"""37        with mock.patch.object(WeChatTag, "iter_tag_users"),\38            mock.patch.object(WeChatUser, "upsert_users"):39            WeChatTag.iter_tag_users.return_value = ["openid1", "openid2"]40            WeChatUser.upsert_users.return_value = ["openid1", "openid2"]41            tag = UserTag.objects.create(42                app=self.app, name="tag", id=101, _tag_local=True)43            del tag._tag_local44            self.assertEqual(tag.sync_users(False), ["openid1", "openid2"])45    def test_change_user_tags(self):46        """æµè¯ç¨æ·æ ç¾åæ´"""47        tag_user_err = "tag_user_err"48        untag_user_err = "untag_user_err"49        user = WeChatUser.objects.create(app=self.app, openid="openid1")50        tag1 = UserTag.objects.create(51            app=self.app, id=101, name="tag1", _tag_local=True)52        del tag1._tag_local53        tag2 = UserTag.objects.create(54            app=self.app, id=102, name="tag2", _tag_local=True)55        del tag2._tag_local56        with mock.patch.object(WeChatTag, "tag_user"),\57            mock.patch.object(WeChatTag, "untag_user"):58            # æµè¯ç¨æ·æ·»å æ ç¾59            WeChatTag.untag_user.side_effect = Exception(untag_user_err)60            user.tags.set((tag1,))61            user.tags.filter(id=tag1.id).get()62            self.assertEqual(WeChatTag.tag_user.call_args, ((tag1.id, user.openid),))63            user.tags.set((tag1, tag2))64            self.assertEqual(user.tags.count(), 2)65            self.assertEqual(66                WeChatTag.tag_user.call_args, ((tag2.id, user.openid),))67            # æµè¯ç¨æ·å åæ ç¾68            WeChatTag.untag_user.side_effect = None69            WeChatTag.tag_user.side_effect = Exception(tag_user_err)70            user.tags.set((tag2, ))71            self.assertEqual(user.tags.count(), 1)72            self.assertEqual(73                WeChatTag.untag_user.call_args, ((tag1.id, user.openid),))74            # æµè¯ç¨æ·å åæ ç¾å¼å¸¸75            WeChatTag.untag_user.side_effect = Exception(untag_user_err)76            with transaction.atomic(), self.assertRaises(Exception) as context:77                user.tags.add(tag1)78            self.assertIn(tag_user_err, str(context.exception))79            self.assertRaises(80                UserTag.DoesNotExist,81                lambda: user.tags.get(app=self.app, id=tag1.id))82            with transaction.atomic(), self.assertRaises(Exception) as context:83                user.tags.remove(tag2)84            self.assertIn(untag_user_err, str(context.exception))85            user.tags.get(app=self.app, id=tag2.id)86    def test_tag_users(self):87        """æµè¯æ ç¾ç¨æ·åæ´"""88        tag_user_err = "tag_user_err"89        untag_user_err = "untag_user_err"90        user1 = WeChatUser.objects.create(app=self.app, openid="openid1")91        user2 = WeChatUser.objects.create(app=self.app, openid="openid2")92        tag = UserTag.objects.create(93            app=self.app, id=101, name="tag1", _tag_local=True)94        del tag._tag_local95        with mock.patch.object(WeChatTag, "tag_user"),\96            mock.patch.object(WeChatTag, "untag_user"):97            # æµè¯æ ç¾æ·»å ç¨æ·98            WeChatTag.untag_user.side_effect = Exception(untag_user_err)99            tag.users.set((user1,))100            tag.users.filter(openid=user1.openid).get()101            self.assertEqual(WeChatTag.tag_user.call_args, ((tag.id, [user1.openid]),))102            tag.users.set((user1, user2))103            self.assertEqual(tag.users.count(), 2)104            self.assertEqual(105                WeChatTag.tag_user.call_args, ((tag.id, [user2.openid]),))106            # æµè¯æ ç¾å åç¨æ·107            WeChatTag.untag_user.side_effect = None108            WeChatTag.tag_user.side_effect = Exception(tag_user_err)109            tag.users.set((user2, ))110            self.assertEqual(tag.users.count(), 1)111            self.assertEqual(112                WeChatTag.untag_user.call_args, ((tag.id, [user1.openid]),))113            # æµè¯æ ç¾å åç¨æ·å¼å¸¸114            WeChatTag.untag_user.side_effect = Exception(untag_user_err)115            with transaction.atomic(), self.assertRaises(Exception) as context:116                tag.users.add(user1)117            self.assertIn(tag_user_err, str(context.exception))118            self.assertRaises(119                WeChatUser.DoesNotExist,120                lambda: tag.users.get(app=self.app, openid=user1.openid))121            with transaction.atomic(), self.assertRaises(Exception) as context:122                tag.users.remove(user2)123            self.assertIn(untag_user_err, str(context.exception))124            tag.users.get(app=self.app, openid=user2.openid)125    def test_edit_tag(self):126        """æµè¯æ ç¾çå¢å æ¹"""127        id = 101128        name = "group1"129        # æ°å¢130        with mock.patch.object(WeChatTag, "create"):131            WeChatTag.create.return_value = dict(132                id=id,133                name=name134            )135            UserTag(app=self.app, name=name).save()136            tag = UserTag.objects.get(app=self.app, name=name)137            self.assertEqual(tag.id, id)138            self.assertEqual(WeChatTag.create.call_args, ((name,),))139        name = "edit"140        with mock.patch.object(WeChatTag, "update"):141            tag.name = name142            tag.save()143            tag = UserTag.objects.get(app=self.app, name=name)144            self.assertEqual(tag.id, id)145            self.assertEqual(WeChatTag.update.call_args, ((id, name),))146        with mock.patch.object(WeChatTag, "delete"):147            tag.delete()148            self.assertRaises(149                UserTag.DoesNotExist,150                lambda: UserTag.objects.get(app=self.app, id=id))151            self.assertEqual(WeChatTag.delete.call_args, ((id,),))152    @property153    def base_tags(self):154        return [155            {156                "id": 0,157                "name": "æªåç»",158                "count": 0159            },160            {161                "id": 1,162                "name": "é»åå",163                "count": 0164            },165            {166                "id": 2,167                "name": "ææ ç»",168                "count": 0169            }...

test_userid.py

Source:test_userid.py

...59        fw.xapi60        fw.userid.batch_start()61        fw.userid.tag_user("user1", ["tag1",])62        fw.userid.tag_user("user2", ["tag1",])63    def test_batch_untag_user(self):64        fw = panos.firewall.Firewall(65            "fw1", "user", "passwd", "authkey", serial="Serial", vsys="vsys2"66        )67        fw.xapi68        fw.userid.batch_start()69        fw.userid.untag_user("user1", ["tag1",])70        fw.userid.untag_user("user2", ["tag1",])71if __name__ == "__main__":...

tagger_sql.py

Source:tagger_sql.py

1import threading2from sqlalchemy import Column, String, Integer3from tg_bot.modules.sql import BASE, SESSION4class Tagger(BASE):5    __tablename__ = "tagger"6    chat_id = Column(String(14), primary_key=True)7    user_id = Column(Integer, primary_key=True)8    def __init__(self, chat_id, user_id):9        self.chat_id = str(chat_id)  # ensure string10        self.user_id = user_id11    def __repr__(self):12        return "<Tag %s>" % self.user_id13Tagger.__table__.create(checkfirst=True)14TAG_INSERTION_LOCK = threading.RLock()15def tag(chat_id, user_id):16    with TAG_INSERTION_LOCK:17        tag_user = Tagger(str(chat_id), user_id)18        SESSION.add(tag_user)19        SESSION.commit()20def is_tag(chat_id, user_id):21    try:22        return SESSION.query(Tagger).get((str(chat_id), user_id))23    finally:24        SESSION.close()25def untag(chat_id, user_id):26    with TAG_INSERTION_LOCK:27        untag_user = SESSION.query(Tagger).get((str(chat_id), user_id))28        if untag_user:29            SESSION.delete(untag_user)30            SESSION.commit()31            return True32        else:33            SESSION.close()34            return False35def tag_list(chat_id):36    try:37        return (SESSION.query(Tagger).filter(38            Tagger.chat_id == str(chat_id)).order_by(39                Tagger.user_id.asc()).all())40    finally:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.