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

How to use _is_active method in localstack

Best Python code snippet using localstack_python

transaction.py

Source:transaction.py

1# ported from:2# https://github.com/aio-libs/aiopg/blob/master/aiopg/sa/transaction.py3from . import exc4class Transaction(object):5    """Represent a database transaction in progress.6    The Transaction object is procured by7    calling the SAConnection.begin() method of8    SAConnection:9        with (yield from engine) as conn:10            trans = yield from conn.begin()11            try:12                yield from conn.execute("insert into x (a, b) values (1, 2)")13            except Exception:14                yield from trans.rollback()15            else:16                yield from trans.commit()17    The object provides .rollback() and .commit()18    methods in order to control transaction boundaries.19    See also:  SAConnection.begin(), SAConnection.begin_twophase(),20    SAConnection.begin_nested().21    """22    def __init__(self, connection, parent):23        self._connection = connection24        self._parent = parent or self25        self._is_active = True26    @property27    def is_active(self):28        """Return ``True`` if a transaction is active."""29        return self._is_active30    @property31    def connection(self):32        """Return transaction's connection (SAConnection instance)."""33        return self._connection34    async def close(self):35        """Close this transaction.36        If this transaction is the base transaction in a begin/commit37        nesting, the transaction will rollback().  Otherwise, the38        method returns.39        This is used to cancel a Transaction without affecting the scope of40        an enclosing transaction.41        """42        if not self._parent._is_active:43            return44        if self._parent is self:45            await self.rollback()46        else:47            self._is_active = False48    async def rollback(self):49        """Roll back this transaction."""50        if not self._parent._is_active:51            return52        await self._do_rollback()53        self._is_active = False54    async def _do_rollback(self):55        await self._parent.rollback()56    async def commit(self):57        """Commit this transaction."""58        if not self._parent._is_active:59            raise exc.InvalidRequestError("This transaction is inactive")60        await self._do_commit()61        self._is_active = False62    async def _do_commit(self):63        pass64    async def __aenter__(self):65        return self66    async def __aexit__(self, exc_type, exc_val, exc_tb):67        if exc_type:68            await self.rollback()69        else:70            if self._is_active:71                await self.commit()72class RootTransaction(Transaction):73    def __init__(self, connection):74        super().__init__(connection, None)75    async def _do_rollback(self):76        await self._connection._rollback_impl()77    async def _do_commit(self):78        await self._connection._commit_impl()79class NestedTransaction(Transaction):80    """Represent a 'nested', or SAVEPOINT transaction.81    A new NestedTransaction object may be procured82    using the SAConnection.begin_nested() method.83    The interface is the same as that of Transaction class.84    """85    _savepoint = None86    def __init__(self, connection, parent):87        super(NestedTransaction, self).__init__(connection, parent)88    async def _do_rollback(self):89        assert self._savepoint is not None, "Broken transaction logic"90        if self._is_active:91            await self._connection._rollback_to_savepoint_impl(92                self._savepoint, self._parent)93    async def _do_commit(self):94        assert self._savepoint is not None, "Broken transaction logic"95        if self._is_active:96            await self._connection._release_savepoint_impl(97                self._savepoint, self._parent)98class TwoPhaseTransaction(Transaction):99    """Represent a two-phase transaction.100    A new TwoPhaseTransaction object may be procured101    using the SAConnection.begin_twophase() method.102    The interface is the same as that of Transaction class103    with the addition of the .prepare() method.104    """105    def __init__(self, connection, xid):106        super().__init__(connection, None)107        self._is_prepared = False108        self._xid = xid109    @property110    def xid(self):111        """Returns twophase transaction id."""112        return self._xid113    async def prepare(self):114        """Prepare this TwoPhaseTransaction.115        After a PREPARE, the transaction can be committed.116        """117        if not self._parent.is_active:118            raise exc.InvalidRequestError("This transaction is inactive")119        await self._connection._prepare_twophase_impl(self._xid)120        self._is_prepared = True121    async def _do_rollback(self):122        await self._connection.rollback_prepared(123            self._xid, is_prepared=self._is_prepared)124    async def _do_commit(self):125        await self._connection.commit_prepared(...

tests.py

Source:tests.py

...8    def setUp(self):9        self.unit = Unit.objects.get(pk=1)10    def test_is_active_dates(self):11        p = Promotion(unit=self.unit, name="test", start_date=datetime(2010,10,2), end_date=datetime(2010,10,5))12        self.assertTrue(p._is_active(datetime(2010,10,3)))13        self.assertFalse(p._is_active(datetime(2010,10,1)))14        self.assertFalse(p._is_active(datetime(2010,10,6)))15    def test_is_active_weekday(self):16        p = Promotion(unit=self.unit, name="test", weekdays='1,2')17        self.assertTrue(p._is_active(datetime(2010,10,4)))18        self.assertTrue(p._is_active(datetime(2010,10,5)))19        self.assertFalse(p._is_active(datetime(2010,10,6)))20    def test_is_active_hours(self):21        p = Promotion(unit=self.unit, name="test", start_hour='10:00', end_hour='12:00')22        self.assertTrue(p._is_active(datetime(2010,10,4,11,0)))23        self.assertFalse(p._is_active(datetime(2010,10,4,9,59)))24        self.assertFalse(p._is_active(datetime(2010,10,4,12,1)))25    def test_is_active_dates_weekdays(self):26        p = Promotion(unit=self.unit, name="test", weekdays="1,2", start_date=datetime(2010,10,2), end_date=datetime(2010,10,5))27        self.assertTrue(p._is_active(datetime(2010,10,4)))28        self.assertFalse(p._is_active(datetime(2010,10,3)))29        self.assertFalse(p._is_active(datetime(2010,10,1)))30        self.assertFalse(p._is_active(datetime(2010,10,11)))31    def test_is_active_dates_hours(self):32        p = Promotion(unit=self.unit, name="test", start_date=datetime(2010,10,2), end_date=datetime(2010,10,5), start_hour='11:30', end_hour='12:15')33        self.assertTrue(p._is_active(datetime(2010,10,4,11,45)))34        self.assertFalse(p._is_active(datetime(2010,10,4,10,0)))35        self.assertFalse(p._is_active(datetime(2010,10,4,12,30)))36        self.assertFalse(p._is_active(datetime(2010,10,1,11,45)))37    def test_is_active_dates_weekdays_hours(self):38        p = Promotion(unit=self.unit, name="test", weekdays="1,2", start_date=datetime(2010,10,2), end_date=datetime(2010,10,5), start_hour='11:30', end_hour='12:15')39        self.assertTrue(p._is_active(datetime(2010,10,4,11,45)))40        self.assertFalse(p._is_active(datetime(2010,10,4,10,0)))41        self.assertFalse(p._is_active(datetime(2010,10,4,12,30)))42        self.assertFalse(p._is_active(datetime(2010,10,3,11,45)))43        self.assertFalse(p._is_active(datetime(2010,10,1,11,45)))...

gcoffee.py

Source:gcoffee.py

1#!/bin/env python32import os3import typing4from pathlib import Path5import gi6gi.require_version("Gtk", "3.0")7from gi.repository import Gtk  # NOQA8class Inhibitor(typing.Protocol):9    def toggle(self):10        ...11class CmdIngibitor:12    on_command = ""13    off_command = ""14    _is_active = False15    def __init__(self):16        self.off()17    def off(self):18        os.system(self.off_command)19        self._is_active = False20    def on(self):21        os.system(self.on_command)22        self._is_active = True23    def toggle(self):24        if self._is_active:25            self.off()26        else:27            self.on()28        print(self.__class__.__name__, " is ", self._is_active)29class DmpsInhibitor(CmdIngibitor):30    on_command = "xset -dpms"31    off_command = "xset +dpms"32class XorgInhibitor(CmdIngibitor):33    on_command = "xset s off"34    off_command = "xset s on"35class App:36    _is_active = False37    def __init__(self, *inhibitors: Inhibitor):38        self._inhibitors = inhibitors39        icon_root = Path(__file__).absolute().parent40        self._off_icon = str(icon_root / "coffee.svg")41        self._on_icon = str(icon_root / "coffee-hot.svg")42        self._icon = Gtk.StatusIcon()43        self._icon.set_from_file(self._off_icon)44        self._icon.connect("activate", self.on_left_click)45    def toggle_icon(self):46        if self._is_active:47            self._icon.set_from_file(self._off_icon)48        else:49            self._icon.set_from_file(self._on_icon)50        self._is_active = not self._is_active51    def on_left_click(self, _):52        for inh in self._inhibitors:53            inh.toggle()54        self.toggle_icon()55    def run(self):56        return Gtk.main()57def main():58    app = App(DmpsInhibitor(), XorgInhibitor())59    app.run()60if __name__ == "__main__":...

handler.py

Source:handler.py

1from datetime import datetime, date2from dateutil import rrule3from sleepy.wake.thread import WakeThread4from sleepy.player import player_factory5from sleepy import config6class WakeHandler:7    _wake_thread = None8    _is_active = False9    def __init__(self):10        self._is_active = False11    def get_wake_time(self):12        c = config.get()13        t = c.get('wake', 'time')14        dtstart = datetime.combine(15            date.today(),16            datetime.strptime(t, '%H:%M').time()17        )18        for wt in rrule.rrule(rrule.DAILY, dtstart=dtstart):19            if wt > datetime.now():20                return wt21    def is_playing(self):22        if self._wake_thread:23            return self._wake_thread.is_playing()24        return False25    def toggle(self):26        if self._is_active:27            self._wake_thread.destroy()28            self._wake_thread.join()29            self._wake_thread = None30            self._is_active = False31        else:32            wake_time = self.get_wake_time()33            player_name = config.get().get('wake', 'player')34            player = player_factory(player_name)35            self._wake_thread = WakeThread(player, wake_time)36            self._wake_thread.start()37            self._is_active = True38    def is_active(self) -> bool:...

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.