semvec/pypi-packages huggingface dataset:
celery_mock 1.1.4
celery-mock allows you to mock celery task and run them when you want
License: MIT
Keywords: celery

===============================
Celery Task Mocking
===============================


.. image:: https://img.shields.io/pypi/v/celery-mock.svg
        :target: https://pypi.python.org/pypi/celery-mock

.. image:: https://img.shields.io/travis/ulamlabs/celery-mock.svg
        :target: https://travis-ci.org/ulamlabs/celery-mock

.. image:: https://codecov.io/gh/ulamlabs/celery-mock/branch/master/graph/badge.svg
  :target: https://codecov.io/gh/ulamlabs/celery-mock


celery-mock allows you to mock celery task and then run them when you want

Rationale
==========

Sometimes a celery task retries itself to wait for some event or model to change.
This is hard to test because celery tasks run (and retry) inline in tests.
Now you can choose when to run your tasks.

Supported versions
==================

- Python 3 support
- Celery 3.1.x and 4.0.x support 


How to install
================

    .. code-block:: bash
    
        pip install celery-mock


How to use
==========

    .. code-block:: python
    
        from celery_mock import task_mock
        from django.test import TestCase, Client
        
        from myapp import dummyview
        
        class UsersTestCase(TestCase):
         
            def test_create_user(self):
                client = Client()
                client.post('/api/users/', data={'username': 'konrad')  # runs tasks inline
                
                with task_mock():
                    client.post('/api/users/', data={'username': 'konrad')
                    # no tasks started yet
                # all tasks ran here
                
                with task_mock('myapp.post_user_create_task'):
                    client.post('/api/users/', data={'username': 'konrad')
                    # all tasks started execept myapp.post_user_create_task
                # myapp.post_user_create_task started here
                
                # you can use task_mock manually:
                
                tmock = task_mock().start()
                client.post('/api/users/', data={'username': 'konrad')
                # no tasks started yet
                tomock.stop()  # all tasks ran here
tr064 0.1.1
TR-064 python API
License: MIT
Requires Python: >=3.6
Keywords: TR-064 LAN DSL CPE configuration broadband forum home router

[![PyPI version](https://img.shields.io/pypi/v/tr064.svg)](https://pypi.org/project/tr064)
[![Python versions](https://img.shields.io/pypi/pyversions/tr064.svg)](https://pypi.org/project/tr064)
[![PyPI - License](https://img.shields.io/pypi/l/tr064)](https://pypi.org/project/tr064)
[![Build Status](https://travis-ci.org/bfueldner/tr064.svg?branch=master)](https://travis-ci.org/bfueldner/tr064)
[![Coverage Status](https://coveralls.io/repos/github/bfueldner/tr064/badge.svg?branch=master)](https://coveralls.io/github/bfueldner/tr064?branch=master)
[![Documentation Status](https://readthedocs.org/projects/tr064/badge/?version=latest)](https://tr064.readthedocs.io/en/latest/?badge=latest)

# TR-064 python API

This package provides a very lightweight and straight-forward TR-064 client, that pulls every information about supported services from the routers description files.

## Installation

Easiest way to install is to use pip:

    $ python3 -m pip install tr064

To update a previous version:

    $ python3 -m pip install --upgrade tr064

## Documentation

Full documentation can be read unter https://tr064.readthedocs.org
apubsub 0.2.5
Message service implementing publisher/subscriber pattern
License: MIT
Requires Python: >=3.7,<4.0

# APubSub

[![Build Status](https://travis-ci.org/outcatcher/apubsub.svg?branch=master)](https://travis-ci.org/outcatcher/apubsub)
[![Coverage](https://codecov.io/gh/outcatcher/apubsub/branch/master/graph/badge.svg)](https://codecov.io/gh/outcatcher/apubsub)
[![PyPI version](https://img.shields.io/pypi/v/apubsub.svg)](https://pypi.org/project/apubsub/)
![GitHub](https://img.shields.io/github/license/outcatcher/apubsub)

Simple, single-purpose message service implementation.

### Installation

_Python versin 3.7+ required_

Just install it with pip: `pip install apubsub`

### Usage


```python
from apubsub import Service

service = Service()

# Note that service is started in stand-alone process
# so start it as early as possible to minimize resource pickling*
service.start()

class Klass:

    def __init__(self):
        self.sub = service.get_client()
        await self.sub.start_consuming()  # subscriber should be started
        
        self.pub = service.get_client()  # if used only as publisher, it is not required

    async def do_smth(self):
        data = await self.sub.get(.1)  # fetch received data with timeout
        if data is None:
            print("No data received by subscriber")
            return
        print(data)
    
    async def do_smth_else(self):
        msg = "some string data"
        await self.pub.publish("topic", msg)

    async def use_iter_get(self):
        async for data in self.sub.get_iter():
            print(f"Data received: {data}")

```

_Check out more examples in tests_