First simple attempts to prevent the "@celery.task is not callable" error as well as initial project structure for tests and installation

Author: carlio

pylint_celery/__init__.py

@@ -0,0 +1,26 @@
+from astroid import MANAGER
+from astroid.builder import AstroidBuilder
+from astroid import nodes
+
+
+MODULE_TRANSFORMS = {}
+
+
+def transform(module):
+    try:
+        tr = MODULE_TRANSFORMS[module.name]
+    except KeyError:
+        pass
+    else:
+        tr(module)
+MANAGER.register_transform(nodes.Module, transform)
+
+
+def celery_transform(module):
+    fake = AstroidBuilder(MANAGER).string_build('''
+class task_dummy(object): pass
+''')
+    module.locals['task'] = fake.locals['task_dummy']
+
+
+MODULE_TRANSFORMS['celery'] = celery_transform

pylint_celery/py2celery.py

@@ -0,0 +1,23 @@
+# -*- coding: UTF-8 -*-
+from distutils.core import setup
+from setuptools import find_packages
+import time
+
+
+_version = "0.1.dev%s" % int(time.time())
+_packages = find_packages(exclude=["*.tests", "*.tests.*", "tests.*", "tests"])
+
+_short_description = "pylint-celery is a Pylint plugin to aid Pylint in recognising and understanding" \
+                     "errors caused when using the Celery library"
+
+
+setup( name='pylint-celery',
+       url='https://github.com/landscapeio/pylint-celery',
+       author='landscape.io',
+       author_email='[email protected]',
+       description=_short_description,
+       version=_version,
+       packages=_packages,
+       license='GPLv2',
+       keywords=('pylint', 'celery', 'plugin')
+)

setup.py

@@ -0,0 +1,11 @@
+"""
+Checks that Pylint does not complain about certain aspects of the Celery library
+"""
+#  pylint: disable=C0111,R0903,W0232
+__revision__ = ''
+
+from celery import task
+
+@task(queue='celery')
+def test_task(an_arg, another_arg):
+    return an_arg + another_arg

test/input/__init__.py

@@ -0,0 +1,28 @@
+
+from pylint_celery import py2celery
+
+from os.path import join, dirname, abspath
+import unittest
+from logilab.common import testlib
+from pylint.testutils import make_tests, LintTestUsingModule, LintTestUsingFile, cb_test_gen
+import sys
+
+
+INPUT_DIR = join(dirname(abspath(__file__)), 'input')
+MESSAGES_DIR = join(dirname(abspath(__file__)), 'messages')
+CALLBACKS = [cb_test_gen(LintTestUsingModule), cb_test_gen(LintTestUsingFile)]
+FILTER_RGX = None
+
+
+def suite():
+    return testlib.TestSuite([unittest.makeSuite(test, suiteClass=testlib.TestSuite)
+                              for test in make_tests(INPUT_DIR, MESSAGES_DIR,
+                                                     FILTER_RGX, CALLBACKS)])
+
+if __name__=='__main__':
+    if len(sys.argv) > 1:
+        FILTER_RGX = sys.argv[1]
+        del sys.argv[1]
+    testlib.unittest_main(defaultTest='suite')
+
+

test/input/func_noerror_celery.py

@@ -0,0 +1,2 @@
+pylint>=1.0.0
+celery>=3.0.0