API tools faq
paste
Guest User

Untitled

a guest
Feb 27th, 2018
126
0
Never
Add comment
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
text 10.78 KB | None | 0 0
raw download clone embed print report
  1. from django.db import models
  2. from django.contrib.auth.models import User
  3. from django.db.models.signals import post_save
  4. from django.dispatch import receiver
  5.  
  6. class Investment(models.Model):
  7. """the main model"""
  8. status = models.IntegerField()
  9.  
  10.  
  11. class InvestmentStatusTrack(models.Model):
  12. """track every change of status on an investment"""
  13. investment = models.ForeignKey(Investment, on_delete=models.CASCADE)
  14. status = models.IntegerField()
  15. modified_on = models.DateTimeField(
  16. blank=True, null=True, default=None, verbose_name=_('modified on'), db_index=True
  17. )
  18. modified_by = models.ForeignKey(
  19. User, blank=True, null=True, default=None, verbose_name=_('modified by'), on_delete=models.CASCADE
  20. )
  21.  
  22. class Meta:
  23. ordering = ('-modified_on', )
  24.  
  25. def __str__(self):
  26. return '{0} - {1}'.format(self.investment, self.status)
  27.  
  28.  
  29. @receiver(post_save, sender=Investment)
  30. def handle_status_track(sender, instance, created, **kwargs):
  31. """add a new track every time the investment status change"""
  32. request = get_request() # a way to get the current request
  33. modified_by = None
  34. if request and request.user and request.user.is_authenticated:
  35. modified_by = request.user
  36. InvestmentStatusTrack.objects.create(
  37. investment=instance, status=instance.status, modified_on=datetime.now(), modified_by=modified_by
  38. )
  39.  
  40. Traceback (most recent call last):
  41. File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 209, in __call__
  42. self._post_teardown()
  43. File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 893, in _post_teardown
  44. self._fixture_teardown()
  45. File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 1041, in _fixture_teardown
  46. connections[db_name].check_constraints()
  47. File "/env/lib/python3.6/site-packages/django/db/backends/postgresql/base.py", line 235, in check_constraints
  48. self.cursor().execute('SET CONSTRAINTS ALL IMMEDIATE')
  49. File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 68, in execute
  50. return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
  51. File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 77, in _execute_with_wrappers
  52. return executor(sql, params, many, context)
  53. File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 85, in _execute
  54. return self.cursor.execute(sql, params)
  55. File "/env/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
  56. raise dj_exc_value.with_traceback(traceback) from exc_value
  57. File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 83, in _execute
  58. return self.cursor.execute(sql)
  59. django.db.utils.IntegrityError: insert or update on table "investments_investmentstatustrack" violates foreign key constraint "investments_investme_modified_by_id_3a12fb21_fk_auth_user"
  60. DETAIL: Key (modified_by_id)=(1) is not present in table "auth_user".
  61.  
  62. class TrackInvestmentStatusTest(ApiTestCase):
  63.  
  64. def login(self, is_staff=False):
  65. password = "abc123"
  66. self.user = mommy.make(User, is_staff=is_staff, is_active=True)
  67. self.user.set_password(password)
  68. self.user.save()
  69. self.assertTrue(self.client.login(username=self.user.username, password=password))
  70.  
  71. def test_add_investment(self):
  72. """it should add a new investment and add a track"""
  73. self.login()
  74.  
  75. url = reverse('investments:investments-list')
  76.  
  77. data = {}
  78.  
  79. response = self.client.post(url, data=data)
  80.  
  81. self.assertEqual(response.status_code, status.HTTP_201_CREATED)
  82.  
  83. self.assertEqual(1, Investment.objects.count())
  84. investment = Investment.objects.all()[0]
  85. self.assertEqual(investment.status, Investment.STATUS_IN_PROJECT)
  86.  
  87. self.assertEqual(1, InvestmentStatusTrack.objects.count())
  88. track = InvestmentStatusTrack.objects.all()[0]
  89. self.assertEqual(track.status, investment.status)
  90. self.assertEqual(track.investment, investment)
  91. self.assertEqual(track.modified_by, self.user)
  92. self.assertEqual(track.modified_on.date(), date.today())
  93.  
  94. def test_save_status(self):
  95. """it should modify the investment and add a track"""
  96.  
  97. self.login()
  98.  
  99. investment_status = Investment.STATUS_IN_PROJECT
  100.  
  101. investment = mommy.make(Investment, asset=asset, status=investment_status)
  102. investment_id = investment.id
  103.  
  104. self.assertEqual(1, InvestmentStatusTrack.objects.count())
  105. track = InvestmentStatusTrack.objects.all()[0]
  106. self.assertEqual(track.status, investment.status)
  107. self.assertEqual(track.investment, investment)
  108. self.assertEqual(track.modified_by, None)
  109. self.assertEqual(track.modified_on.date(), date.today())
  110.  
  111. url = reverse('investments:investments-detail', args=[investment.id])
  112.  
  113. data = {
  114. 'status': Investment.STATUS_ACCEPTED
  115. }
  116.  
  117. response = self.client.patch(url, data=data)
  118.  
  119. self.assertEqual(response.status_code, status.HTTP_200_OK)
  120.  
  121. self.assertEqual(1, Investment.objects.count())
  122. investment = Investment.objects.all()[0]
  123. self.assertEqual(investment.id, investment_id)
  124. self.assertEqual(investment.status, Investment.STATUS_ACCEPTED)
  125.  
  126. self.assertEqual(2, InvestmentStatusTrack.objects.count())
  127. track = InvestmentStatusTrack.objects.all()[0]
  128. self.assertEqual(track.status, Investment.STATUS_ACCEPTED)
  129. self.assertEqual(track.investment, investment)
  130. self.assertEqual(track.modified_by, self.user)
  131. self.assertEqual(track.modified_on.date(), date.today())
  132.  
  133. track = InvestmentStatusTrack.objects.all()[1]
  134. self.assertEqual(track.status, Investment.STATUS_IN_PROJECT)
  135. self.assertEqual(track.investment, investment)
  136. self.assertEqual(track.modified_by, None)
  137. self.assertEqual(track.modified_on.date(), date.today())
  138.  
  139. response = self.client.post(url, data=data)
  140.  
  141. investment = mommy.make(Investment, status=investment_status)
  142.  
  143. def set_user(user):
  144. current_request = get_request()
  145.  
  146. if current_request:
  147. current_request.user = user
  148.  
  149. def login(self, is_staff=False):
  150. password = "abc123"
  151. self.user = mommy.make(User, is_staff=is_staff, is_active=True)
  152. self.user.set_password(password)
  153. self.user.save()
  154. self.assertTrue(self.client.login(username=self.user.username, password=password))
  155. set_user(self.user)
  156.  
  157. File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 1041, in _fixture_teardown
  158. connections[db_name].check_constraints()
  159.  
  160. def _fixture_teardown(self):
  161. if not connections_support_transactions():
  162. return super()._fixture_teardown()
  163. try:
  164. for db_name in reversed(self._databases_names()):
  165. if self._should_check_constraints(connections[db_name]):
  166. connections[db_name].check_constraints()
  167. finally:
  168. self._rollback_atomics(self.atomics)
  169.  
  170. def __call__(self, result=None):
  171. """
  172. Wrapper around default __call__ method to perform common Django test
  173. set up. This means that user-defined Test Cases aren't required to
  174. include a call to super().setUp().
  175. """
  176. testMethod = getattr(self, self._testMethodName)
  177. skipped = (
  178. getattr(self.__class__, "__unittest_skip__", False) or
  179. getattr(testMethod, "__unittest_skip__", False)
  180. )
  181.  
  182. if not skipped:
  183. try:
  184. self._pre_setup()
  185. except Exception:
  186. result.addError(self, sys.exc_info())
  187. return
  188. super().__call__(result)
  189. if not skipped:
  190. try:
  191. self._post_teardown()
  192. except Exception:
  193. result.addError(self, sys.exc_info())
  194. return
  195.  
  196. class Investment(models.Model):
  197. """the main model"""
  198. status = models.IntegerField()
  199.  
  200. def handle_status_track(self):
  201. """add a new track every time the investment status change"""
  202. request = get_request() # a way to get the current request
  203. modified_by = None
  204. if request and request.user and request.user.is_authenticated:
  205. modified_by = request.user
  206. InvestmentStatusTrack.objects.create(
  207. investment=self, status=self.status, modified_on=datetime.now(), modified_by=modified_by
  208. )
  209.  
  210.  
  211. class InvestmentStatusTrack(models.Model):
  212. """track every change of status on an investment"""
  213. investment = models.ForeignKey(Investment, on_delete=models.CASCADE)
  214. status = models.IntegerField()
  215. modified_on = models.DateTimeField(
  216. blank=True, null=True, default=None, verbose_name=_('modified on'), db_index=True
  217. )
  218. modified_by = models.ForeignKey(
  219. User, blank=True, null=True, default=None, verbose_name=_('modified by'), on_delete=models.CASCADE
  220. )
  221.  
  222. class Meta:
  223. ordering = ('-modified_on',)
  224.  
  225. class InvestmentViewSet(ViewSet):
  226. model = Investment
  227. serializer_class = InvestmentSerializer
  228.  
  229. def perform_create(self, serializer):
  230. """save"""
  231. investment = serializer.save()
  232. investment.handle_status_track()
  233.  
  234. def perform_update(self, serializer):
  235. """save"""
  236. investment = serializer.save()
  237. investment.handle_status_track()
  238.  
  239. from threading import current_thread
  240.  
  241. class RequestManager(object):
  242. """get django request from anywhere"""
  243. _shared = {}
  244.  
  245. def __init__(self):
  246. """This is a Borg"""
  247. self.__dict__ = RequestManager._shared
  248.  
  249. def _get_request_dict(self):
  250. """request dict"""
  251. if not hasattr(self, '_request'):
  252. self._request = {} # pylint: disable=attribute-defined-outside-init
  253. return self._request
  254.  
  255. def clean(self):
  256. """clean"""
  257. if hasattr(self, '_request'):
  258. del self._request
  259.  
  260. def get_request(self):
  261. """return request"""
  262. _requests = self._get_request_dict()
  263. the_thread = current_thread()
  264. if the_thread not in _requests:
  265. return None
  266. return _requests[the_thread]
  267.  
  268. def set_request(self, request):
  269. """set request"""
  270. _requests = self._get_request_dict()
  271. _requests[current_thread()] = request
  272.  
  273.  
  274. class RequestMiddleware:
  275. def __init__(self, get_response):
  276. self.get_response = get_response
  277.  
  278. def __call__(self, request):
  279. # Set the request
  280. RequestManager().set_request(request)
  281.  
  282. response = self.get_response(request)
  283.  
  284. # ---- THIS WAS THE MISSING PART -----
  285. # Clear the request
  286. RequestManager().set_request(None)
  287. # ------------------------------------
  288.  
  289. return response
  290.  
  291. def process_exception(self, request, exception):
  292. """handle exceptions"""
  293. # clear request also in case of exception
  294. RequestManager().set_request(None)
  295.  
  296.  
  297. def get_request():
  298. """get current request from anywhere"""
  299. return RequestManager().get_request()
Add Comment
Please, Sign In to add comment
Public Pastes
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy.  OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!