Automatic Imports of Events to Open Event from online event sites with Query Server and Event Collect

One goal for the next version of the Open Event project is to allow an automatic import of events from various event listing sites. We will implement this using Open Event Import APIs and two additional modules: Query Server and Event Collect. The idea is to run the modules as micro-services or as stand-alone solutions.

Query Server
The query server is, as the name suggests, a query processor. As we are moving towards an API-centric approach for the server, query-server also has API endpoints (v1). Using this API you can get the data from the server in the mentioned format. The API itself is quite intuitive.

API to get data from query-server

GET /api/v1/search/<search-engine>/query=query&format=format

Sample Response Header

 Cache-Control: no-cache
 Connection: keep-alive
 Content-Length: 1395
 Content-Type: application/xml; charset=utf-8
 Date: Wed, 24 May 2017 08:33:42 GMT
 Server: Werkzeug/0.12.1 Python/2.7.13
 Via: 1.1 vegur

The server is built in Flask. The GitHub repository of the server contains a simple Bootstrap front-end, which is used as a testing ground for results. The query string calls the search engine result scraper scraper.py that is based on the scraper at searss. This scraper takes search engine, presently Google, Bing, DuckDuckGo and Yahoo as additional input and searches on that search engine. The output from the scraper, which can be in XML or in JSON depending on the API parameters is returned, while the search query is stored into MongoDB database with the query string indexing. This is done keeping in mind the capabilities to be added in order to use Kibana analyzing tools.

The frontend prettifies results with the help of PrismJS. The query-server will be used for initial listing of events from different search engines. This will be accessed through the following API.

The query server app can be accessed on heroku.

➢ api/list​: To provide with an initial list of events (titles and links) to be displayed on Open Event search results.

When an event is searched on Open Event, the query is passed on to query-server where a search is made by calling scraper.py with appending some details for better event hunting. Recent developments with Google include their event search feature. In the Google search app, event searches take over when Google detects that a user is looking for an event.

The feed from the scraper is parsed for events inside query server to generate a list containing Event Titles and Links. Each event in this list is then searched for in the database to check if it exists already. We will be using elastic search to achieve fuzzy searching for events in Open Event database as elastic search is planned for the API to be used.

One example of what we wish to achieve by implementing this type of search in the database follows. The user may search for

-Google Cloud Event Delhi
-Google Event, Delhi
-Google Cloud, Delhi
-google cloud delhi
-Google Cloud Onboard Delhi
-Google Delhi Cloud event

All these searches should match with “Google Cloud Onboard Event, Delhi” with good accuracy. After removing duplicates and events which already exist in the database from this list have been deleted, each event is rendered on search frontend of Open Event as a separate event. The user can click on any of these event, which will make a call to event collect.

Event Collect

The event collect project is developed as a separate module which has two parts

● Site specific scrapers
In its present state, event collect has scrapers for eventbrite and ticket-leap which, given a query, scrape eventbrite (and ticket-leap respectively) search results and downloads JSON files of each event using Loklak‘s API.
The scrapers can be developed in any form or any number of scrapers/scraping tools can be added as long as they are in alignment with the Open Event Import API’s data format. Writing tests for these against the concurrent API formats will take care of this. This part will be covered by using a json-validator​ to check against a pre-generated schema.

● REST APIs
The scrapers are exposed through a set of APIs, which will include, but not limited to,
➢ api/fetch-event : ​to scrape any event given the link and compose the data in a predefined JSON format which will be generated based on Open Event Import API. When this function is called on an event link, scrapers are invoked which collect event data such as event, meta, forms etc. This data will be validated against the generated JSON schema. The scraped JSON and directory structure for media files:
➢ api/export : to export all the JSON data containing event information into Open Event Server. As and when the scraping is complete, the data will be added into Open Event’s database as a new event.

How the Import works

The following graphic shows how the import works.




Let’s dive into the workflow. So as the diagram illustrates, the ‘search​’ functionality makes a call to api/list API endpoint provided by query-server which returns with events’ ‘Title’ and ‘Event Link’ from the parsed XML/JSON feed. This list is displayed as Open Event’s search results. Now the results having been displayed, the user can click on any of the events. When the user clicks on any event, the event is searched for in Open Event’s database. Two things happen now:

  • The event page loads if the event is found.
  • If the event does not already exist in the database, clicking on any event will

➢ Insert this event’s title and link in the database and get the event_id

➢ Make a call to api/fetch-event in event-collect which then invokes a site-specific scraper to fetch data about the event the user has chosen

➢ When the data is scraped, it is imported into Open Event database using the previously generated event_id. The page will be loaded using jquery ajax ​as and when the scraping is done.​When the imports are done, the search page refreshes with the new results. The Open Event Orga Server exposes a well documented REST API that can be used by external services to access the data.

DetachedInstanceError: dealing with Celery, Flask’s app context and SQLAlchemy

In the open event server project, we had chosen to go with celery for async background tasks. From the official website,

What is celery?

Celery is an asynchronous task queue/job queue based on distributed message passing.

What are tasks?

The execution units, called tasks, are executed concurrently on a single or more worker servers using multiprocessing.

After the tasks had been set up, an error constantly came up whenever a task was called

The error was:

DetachedInstanceError: Instance <User at 0x7f358a4e9550> is not bound to a Session; attribute refresh operation cannot proceed

The above error usually occurs when you try to access the session object after it has been closed. It may have been closed by an explicit session.close() call or after committing the session with session.commit().

The celery tasks in question were performing some database operations. So the first thought was that maybe these operations might be causing the error. To test this theory, the celery task was changed to :

@celery.task(name='lorem.ipsum')
def lorem_ipsum():
    pass

But sadly, the error still remained. This proves that the celery task was just fine and the session was being closed whenever the celery task was called. The method in which the celery task was being called was of the following form:

def restore_session(session_id):
    session = DataGetter.get_session(session_id)
    session.deleted_at = None
    lorem_ipsum.delay()
    save_to_db(session, "Session restored from Trash")
    update_version(session.event_id, False, 'sessions_ver')


In our app, the app_context was not being passed whenever a celery task was initiated. Thus, the celery task, whenever called, closed the previous app_context eventually closing the session along with it. The solution to this error would be to follow the pattern as suggested on http://flask.pocoo.org/docs/0.12/patterns/celery/.

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    task_base = celery.Task

    class ContextTask(task_base):
        abstract = True

        def __call__(self, *args, **kwargs):
            if current_app.config['TESTING']:
                with app.test_request_context():
                    return task_base.__call__(self, *args, **kwargs)
            with app.app_context():
                return task_base.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery

celery = make_celery(current_app)


The __call__ method ensures that celery task is provided with proper app context to work with.

 

Event-driven programming in Flask with Blinker signals

Setting up blinker:

The Open Event Project offers event managers a platform to organize all kinds of events including concerts, conferences, summits and regular meetups. In the server part of the project, the issue at hand was to perform multiple tasks in background (we use celery for this) whenever some changes occurred within the event, or the speakers/sessions associated with the event.

The usual approach to this would be applying a function call after any relevant changes are made. But the statements making these changes were distributed all over the project at multiple places. It would be cumbersome to add 3-4 function calls (which are irrelevant to the function they are being executed) in so may places. Moreover, the code would get unstructured with this and it would be really hard to maintain this code over time.

That’s when signals came to our rescue. From Flask 0.6, there is integrated support for signalling in Flask, refer http://flask.pocoo.org/docs/latest/signals/ . The Blinker library is used here to implement signals. If you’re coming from some other language, signals are analogous to events.

Given below is the code to create named signals in a custom namespace:


from blinker import Namespace

event_signals = Namespace()
speakers_modified = event_signals.signal('event_json_modified')

If you want to emit a signal, you can do so by calling the send() method:


speakers_modified.send(current_app._get_current_object(), event_id=event.id, speaker_id=speaker.id)

From the user guide itself:

“ Try to always pick a good sender. If you have a class that is emitting a signal, pass self as sender. If you are emitting a signal from a random function, you can pass current_app._get_current_object() as sender. “

To subscribe to a signal, blinker provides neat decorator based signal subscriptions.


@speakers_modified.connect
def name_of_signal_handler(app, **kwargs):

 

Some Design Decisions:

When sending the signal, the signal may be sending lots of information, which your signal may or may not want. e.g when you have multiple subscribers listening to the same signal. Some of the information sent by the signal may not be of use to your specific function. Thus we decided to enforce the pattern below to ensure flexibility throughout the project.


@speakers_modified.connect
def new_handler(app, **kwargs):
# do whatever you want to do with kwargs['event_id']

In this case, the function new_handler needs to perform some task solely based on the event_id. If the function was of the form def new_handler(app, event_id), an error would be raised by the app. A big plus of this approach, if you want to send some more info with the signal, for the sake of example, if you also want to send speaker_name along with the signal, this pattern ensures that no error is raised by any of the subscribers defined before this change was made.

When to use signals and when not ?

The call to send a signal will of course be lying in another function itself. The signal and the function should be independent of each other. If the task done by any of the signal subscribers, even remotely affects your current function, a signal shouldn’t be used, use a function call instead.

How to turn off signals while testing?

When in testing mode, signals may slow down your testing as unnecessary signals subscribers which are completely independent from the function being tested will be executed numerous times. To turn off executing the signal subscribers, you have to make a small change in the send function of the blinker library.

Below is what we have done. The approach to turn it off may differ from project to project as the method of testing differs. Refer https://github.com/jek/blinker/blob/master/blinker/base.py#L241 for the original function.


def new_send(self, *sender, **kwargs):
    if len(sender) == 0:
        sender = None
    elif len(sender) > 1:
        raise TypeError('send() accepts only one positional argument, '
                        '%s given' % len(sender))
    else:
        sender = sender[0]
    # only this line was changed
    if not self.receivers or app.config['TESTING']:
        return []
    else:
        return [(receiver, receiver(sender, **kwargs))
                for receiver in self.receivers_for(sender)]
                
Signal.send = new_send

event_signals = Namespace
# and so on ....

That’s all for now. Have some fun signaling 😉 .

 

Adding Google Analytics To All Pages Using Flask

Google Analytics gives a detailed insight about your website including how many people visited, time, demography, how many returning visitors and all such information. It’s a real important tool to have. All you have to do is create a Universal Analytics Tracking code and use it in a javascript code. The only problem is this code needs to be present in all the pages that you wants the analytics data for. So changing any part of the javascript code anytime, needs to be changed in all .html files.

However, there is a better way of doing it in flask. Create a file base.html and write the code:

<script>
(function(i,s,o,g,r,a,m){i[‘GoogleAnalyticsObject’]=r;i[r]=i[r]||function(){
(i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o),
m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m)
})(window,document,’script’,’https://www.google-analytics.com/analytics.js’,’ga’);

ga(‘create’, ‘<track-code>’, ‘auto’);
ga(‘send’, ‘pageview’);

</script>

Then using the property of jinja2 template extend this file in all the html files, i.e. {% extends ‘gentelella/admin/base.html’ %}. Thus now when you make some change in the above mentioned javascrpt code, you need to change it only in one place and it is changed in all other places.

R14 – Memory Quota Exceeded

We, like many other organisations, are using heroku as the deployment server for our project open event organizer server. Things are pretty simple and awesome when your project is in its beginning phase and things run pretty smoothly. But as your project grows, there comes some server problem. And one of the biggest problems as your project grows is memory. Now since various packages have a different amount of memory assigned to you in case of hosting in generic servers such as heroku, so it might result in memory quota exceeded. Recently, we faced such a problem. R14 – Memory Quota Exceeded. Took us quite some time to understand what and why and how this occurred. So let me share a few things I found about this error.

Continue reading R14 – Memory Quota Exceeded

Python code examples

I’ve met many weird examples of  behaviour in python language while working on Open Event project. Today I’d like to share some examples with you. I think this knowledge is necessary, if you’d like to increase a  bit your knowledge in python area.

Simple adding one element to python list:

def foo(value, x=[]):
  x.append(value)
  return x

>>> print(foo(1))
>>> print(foo(2))
>>> print(foo(3, []))
>>> print(foo(4))

OUTPUT

[1] 
[1, 2] 
[3]
[1, 2, 4]

First output is obvious, but second not exactly. Let me explain it, It happens because x(empty list) argument is only evaluated once, So on every call foo(), we modify that list, appending a value to it. Finally we have [1,2, 4] output. I recommend to avoid mutable params as default.

Another example:

Do you know which type it is?

>>> print(type([ el for el in range(10)]))
>>> print(type({ el for el in range(10)}))
>>> print(type(( el for el in range(10))))

Again first and second type are obvious <class ‘list’>, <class ‘set’>. You may  think that last one should return type tuple but it returns a generator <class ‘generator’>.

Example:

Do you think that below code returns an exception?

list= [1,2,3,4,5]
>>> print(list [8:])

If you think that above expression returns index error you’re wrong. It returns empty list [].

Example funny boolean operators

>>> 'c' == ('c' or 'b')
True
>>> 'd' == ('a' or 'd')
False
>>> 'c' == ('c' and 'b')
False 
>>> 'd' == ('a' and 'd')
True

You can think that that OR and AND operators are broken.

You have to know how python interpreter behaves while looking for OR and AND operators.

So OR Expression takes the first statement and checks if it is true. If the first statement is true, then Python returns object’s value without checking second value. If first statement is false interpreter checks second value and returns that value.

AND operator checks if first statement is false, the whole statement has to be false. So it returns first value, but if first statement is true it checks second statement and returns second value.

Below i will show you how it works

>>> 'c' == ('c' or 'b')
>>> 'c' == 'c'
True
>>> 'd' == ('a' or 'd')
>>> 'd' == 'a'
False
>>> 'c' == ('c' and 'b')
>>> 'c' == 'b'
False 
>>> 'd' == ('a' and 'd')
>>> 'd' == 'd'
True

I hope that i have explained you how the python interpreter checks OR and AND operators. So know above examples should be more understandable.

GET and POST requests

If you wonder how to get or update page resource, you have to read this article.

It’s trivial if you have basic knowledge about HTTP protocol. I’d like to get you little involved to this subject.

So GET and POST are most useful methods in HTTP protocol.

What is HTTP?

Hypertext transfer protocol – allow us to communicate between client and server side. In Open Event project we use web browser as client and for now we use Heroku for server side.

Difference between GET and POST methods

GET – it allows to get data from specified resources

POST – it allows to submit new data to specified resources for example by html form.

GET samples:

For example we use it to get details about event

curl http://open-event-dev.herokuapp.com/api/v2/events/95

Response from server:

Of course you can use this for another needs, If you are a poker player I suppose that you’d like to know how many percentage you have on hand.

curl http://www.propokertools.com/simulations/show?g=he&s=generic&b&d&h1=AA&h2=KK&h3&h4&h5&h6&_

POST samples:

curl -X POST https://example.com/resource.cgi

You can often find this action in a contact page or in a login page.

How does request look in python?

We use Requests library to communication between client and server side. It’s very readable for developers. You can find great documentation  and a lot of code samples on their website. It’s very important to see how it works.

>>> r = requests.get('https://api.github.com/user', auth=('user', 'pass'))
>>> r.status_code
200

I know that samples are very important, but take a look how Requests library fulfils our requirements in 100%. We have decided to use it because we would like to communicate between android app generator and orga server application. We have needed to send request with params(email, app_name, and api of event url) by post method to android generator resource. It executes the process of sending an email – a package of android application to a provided email address.

data = {
    "email": login.current_user.email,
    "app_name": self.app_name,
    "endpoint": request.url_root + "api/v2/events/" + str(self.event.id)
}
r = requests.post(self.app_link, json=data)

 

Dynamically marshalling output in Flask Restplus

Do you use Flask-Restplus ? Have you felt the need of dynamically modifying API output according to condition. If yes, then this post is for you.

In this post, I will show how to use decorators to restrict GET API output. So let’s start.

This is the basic code to create an API. Here we have created a get_speaker API to get a single item from Speaker model.

from flask_restplus import Resource, Model, fields, Namespace
from models import Speaker

api = Namespace('speakers', description='Speakers', path='/')

SPEAKER = Model('Name', {
	'id': fields.Integer(),
	'name': fields.String(),
	'phone': fields.String()
})

class DAO:
	def get(speaker_id):
		return Speaker.query.get(speaker_id)

@api.route('/speakers/<int:speaker_id>')
class Speaker(Resource):
    @api.doc('get_speaker')
    @api.marshal_with(SPEAKER)
    def get(self, speaker_id):
        """Fetch a speaker given its id"""
        return DAO.get(speaker_id)

Now our need is to change the returned API data according to some condition. Like if user is authenticated then only return phone field of the SPEAKER model. One way to do this is to create condition statements in get method that marshals the output according to the situation. But if there are lots of methods which require this, then this is not a good way.

So let’s create a decorator which can change the marshal decorator at runtime. It will accept parameters as which models to marshal in case of authenticated and non-authenticated cases.

from flask_login import current_user
from flask_restplus import marshal_with

def selective_marshal_with(fields, fields_private):
    """
    Selective response marshalling. Doesn't update apidoc.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if current_user.is_authenticated:
                model = fields
            else:
                model = fields_private
            func2 = marshal_with(model)(func)
            return func2(*args, **kwargs)
        return wrapper
    return decorator

The above code adds a wrapper over the API function which checks if the user is authenticated. If the user is authenticated, fields model is used for marshalling else fields_private is used for marshalling.

So let’s create the private model for SPEAKER. We will call it SPEAKER_PRIVATE.

from flask_restplus import Model, fields

SPEAKER_PRIVATE = Model('NamePrivate', {
	'id': fields.Integer(),
	'name': fields.String()
})

The final step is attaching the selective_marshal_with decorator to the get() method.

@api.route('/speakers/<int:speaker_id>')
class Speaker(Resource):
    @api.doc('get_speaker', model=SPEAKER)
    @selective_marshal_with(SPEAKER, SPEAKER_PRIVATE)
    def get(self, speaker_id):
        """Fetch a speaker given its id"""
        return DAO.get(speaker_id)

You will notice that I removed @api.marshal_with(SPEAKER). This was to disable automatic marshalling of output by flask-restplus. To compensate for this, I have added model=SPEAKER inapi.doc. It will not auto-marshal the output but will still show the swagger documentation.

That concludes this. The get method will now switch marshal field w.r.t to the authentication level of the user. As you may notice, the selective_marhsal_with function is generic and can be used with other models and APIs too.

 

{{ Repost from my personal blog http://aviaryan.in/blog/gsoc/dynamic-marshal-restplus.html }}

Flask-SocketIO Notifications

In the previous post I explained about configuring Flask-SocketIO, Nginx and Gunicorn. This post includes integrating Flask-SocketIO library to display notifications to users in real time.

Flask Config

For development we use the default web server that ships with Flask. For this, Flask-SocketIO fallsback to long-polling as its transport mechanism, instead of WebSockets. So to properly test SocketIO I wanted to work directly with Gunicorn (hence the previous post about configuring development environment). Also, not everyone needs to be bothered with the changes required to run it.

class DevelopmentConfig(Config):
    DEVELOPMENT = True
    DEBUG = True

    # If Env Var `INTEGRATE_SOCKETIO` is set to 'true', then integrate SocketIO
    socketio_integration = os.environ.get('INTEGRATE_SOCKETIO')
    if socketio_integration == 'true':
        INTEGRATE_SOCKETIO = True
    else:
        INTEGRATE_SOCKETIO = False

    # Other stuff

SocketIO is integrated (in development env) if the developer has set the INTEGRATE_SOCKETIO environment variable to “true”. In Production, our application runs on Gunicorn, and SocketIO integration must always be there.

Flow

To send message to a particular connection (or a set of connections) Flask-SocketIO provides Rooms. The connections are made to join a room and the message is sent in the room. So to send message to a particular user we need him to join a room, and then send the message in that room. The room name needs to be unique and related to just one user. The User database Ids could be used. I decided to keep user_{id} as the room name for a user with id {id}. This information (room name) would be needed when making the user join a room, so I stored it for every user that logged in.

@expose('/login/', methods=('GET', 'POST'))
    def login_view(self):
        if request.method == 'GET':
            # Render template
        if request.method == 'POST':
            # Take email and password from form and check if 
            # user exists. If he does, log him in.
            login.login_user(user)

            # Store user_id in session for socketio use
            session['user_id'] = login.current_user.id

            # Redirect

After the user logs in, a connection request from the client is sent to the server. With this connection request the connection handler at server makes the user join a room (based on the user_id stored previously).

@socketio.on('connect', namespace='/notifs')
def connect_handler():
    if current_user.is_authenticated():
        user_room = 'user_{}'.format(session['user_id'])
        join_room(user_room)
        emit('response', {'meta': 'WS connected'})

The client side is somewhat similar to this:

<script src="{{ url_for('static', filename='path/to/socket.io-client/socket.io.js') }}"></script>
<script type="text/javascript">
$(document).ready(function() {
    var namespace = '/notifs';

    var socket = io.connect(location.protocol + "//" + location.host + namespace, {reconnection: false});

    socket.on('response', function(msg) {
        console.log(msg.meta);
        // If `msg` is a notification, display it to the user.
    });
});
</script>

Namespaces helps when making multiple connections over the same socket.

So now that the user has joined a room we can send him notifications. The notification data sent to the client should be standard, so the message always has the same format. I defined a get_unread_notifs method for the User class that fetches unread notifications.

class User(db.Model):
    # Other stuff

    def get_unread_notifs(self, reverse=False):
        """Get unread notifications with titles, humanized receiving time
        and Mark-as-read links.
        """
        notifs = []
        unread_notifs = Notification.query.filter_by(user=self, has_read=False)
        for notif in unread_notifs:
            notifs.append({
                'title': notif.title,
                'received_at': humanize.naturaltime(datetime.now() - notif.received_at),
                'mark_read': url_for('profile.mark_notification_as_read', notification_id=notif.id)
            })

        if reverse:
            return list(reversed(notifs))
        else:
            return notifs

This class method is used when a notification is added in the database and has to be pushed into the user SocketIO room.

def create_user_notification(user, action, title, message):
    """
    Create a User Notification
    :param user: User object to send the notification to
    :param action: Action being performed
    :param title: The message title
    :param message: Message
    """
    notification = Notification(user=user,
                                action=action,
                                title=title,
                                message=message,
                                received_at=datetime.now())
    saved = save_to_db(notification, 'User notification saved')

    if saved:
        push_user_notification(user)

def push_user_notification(user):
    """
    Push user notification to user socket connection.
    """
    user_room = 'user_{}'.format(user.id)
    emit('response',
         {'meta': 'New notifications',
          'notif_count': user.get_unread_notif_count(),
          'notifs': user.get_unread_notifs()},
         room=user_room,
         namespace='/notifs')

Unit Testing

There are many stories about unit testing. Developers sometimes say that they don’t write tests because they write a good quality code. Does it make sense, if no one is infallible?.

At studies only a  few teachers talk about unit testing, but they only show basic examples of unit testing. They require to write a few tests to finish final project, but nobody really  teaches us the importance of unit testing.

I have also always wondered what benefits can it bring. As time is a really important factor in our work it often happens that we simply resign of this part of process development to get “more time” rather than spend time on writing stupid tests. But now I know that it is a vicious circle.

Customers requierments does not help us. They put a high pressure to see visible results not a few statistics about coverage status. None of them cares about some strange numbers. So, as I mentioned above, we usually focuses on building new features and get riid of tests. It may seem to save time, but it doesn’t.

In reality tests save us a lot of time because we can identify and fix bugs very quickly. If a bug ocurrs because someone’s change we don’t have to spend long hours trying to figure out wgat is going out. That’s why we need tests.  

It is especially visible in huge open source projects. FOSSASIA organization has about 200 contributors. In OpenEvent project we have about 20 active developers, who generate many lines of code every single day. Many of them change over and over again as well as interfere  with each other.

Let me provide you with a simple example. In our team we have about 7 pull requests per day. As I mentioned above we want to make our code high quality and free of bugs, but without testing identifying if pull request causes a bug is very difficult task. But fortunately this boring job makes Travis CI for us. It is a great tool which uses our tests and runs them on every PR  to check if bugs occur. It helps us to quickly notice bugs and maintain our project very well.

What is unit testing?

Unit testing is a software development method in which the smallest testable parts of an application are tested

Why do we need writing unit tests?

Let me point all arguments why unit testing is really important while developing a project.

  • To prove that our code works properly

If developer adds another condition, test checks if method returns correct results. You simply don’t need to wonder if something is wrong with you code.

  • To reduce amount of bugs

It let you to know what inputs params’ function should get and what results should be returned. You simply don’t  write unused code

  • To save development time

Developers don’t waste time on checking every code’s change if his code works correctly

  • Unit tests help to understand software design
  • To provide quick feedback about method which you are testing
  • To help document a code

How to write unit test in Python

In my work I write use tests in Python. I am going to share my sample code  with you now

  • Import module unittest
  • Choose function to test
  • Write unit test

Example OpenEvent test in Python

class TestPagesUrls(OpenEventTestCase):

   def setUp(self):

       self.app = Setup.create_app()

   def test_if_urls_exist(self):

       """Test all urls via GET method"""

       with app.test_request_context():

           for rule in app.url_map.iter_rules():

               if excluded_paths(rule):

                   status_code = self.app.get(request.url[:-1] + str(rule).replace('//', '/'),        follow_redirects=True).status_code

                   self.assertTrue(status_code in [200, 302, 401])

 

I want to check if all views exist but it required a lot of time. That’s why I wonder I how to avoid writing similar tests. Finally, based  on our list of routes I am able to write test which checks code’s status  on every page.

If some of them response returns status_code different than 200, 302 or 401, test fails.This results means that somethings is wrong. Simple, isn’t it ?  Try to test it manually…. This one short test cover about 40 use cases…

This example shows an incredible value of unit tests! If developer makes a bug in response he receives an error that something is wrong with a view. Travis CI allows to reject all  wrong pull requests and merge only these which fulfill our quality requirements.   

Fixing  error is one part but finding a bug is even harder task. But an ability to detect bug on early stage of process development reduces cost of software.