Changeset - 790da8154ef8
Parent rev.
Child rev.
[Not reviewed]
default
0 8 0
Mads Kiilerich - 11 years ago 2015-04-28 16:55:05
madski@unity3d.com
tests: minimize the amount of references to rc and the past
8 files changed with 42 insertions and 42 deletions:
kallithea/tests/fixtures/journal_dump.csv
3
3
kallithea/tests/functional/test_login.py
3
3
kallithea/tests/other/manual_test_vcs_operations.py
1
1
kallithea/tests/other/test_libs.py
20
20
kallithea/tests/scripts/create_rc.sh
4
4
kallithea/tests/scripts/test_concurency.py
2
2
kallithea/tests/scripts/test_crawler.py
1
1
kallithea/tests/vcs/test_utils.py
8
8
0 comments (0 inline, 0 general)
kallithea/tests/fixtures/journal_dump.csv
➞
Show inline comments
 
@@ -1689,195 +1689,195 @@ user_log_id,user_id,username,repository_
 
1750,3,demo,333,IAmTesting_this,"",started_following_repo,2012-10-23 19:15:24.347861
 
1751,3,demo,333,IAmTesting_this,"",user_created_repo,2012-10-23 19:15:24.361363
 
1752,3,demo,333,IAmTesting_this,"",user_updated_repo,2012-10-23 19:35:36.723229
 
1753,1,default,333,IAmTesting_this,194.117.18.103,push:31ca1f206de23b0d257dcbb323f9ba94e142ceb5,2012-10-23 19:59:07.360826
 
1754,3,demo,334,Test1qwerty,"",started_following_repo,2012-10-23 20:56:13.001385
 
1755,3,demo,334,Test1qwerty,"",user_created_repo,2012-10-23 20:56:13.014633
 
1756,3,demo,334,Test1qwerty,"",push_local:5957aa6f6f14241663b8b1a0ca050e43c3d3ca27,2012-10-23 20:57:25.88513
 
1757,3,demo,335,git,"",started_following_repo,2012-10-23 21:49:35.974797
 
1758,3,demo,335,git,"",user_created_repo,2012-10-23 21:49:35.994798
 
1759,3,demo,336,fork-cruel-test-2,"",started_following_repo,2012-10-23 23:49:23.770508
 
1760,3,demo,174,releases/cruel-test-2,"",user_forked_repo:fork-cruel-test-2,2012-10-23 23:49:23.85229
 
1761,3,demo,336,fork-cruel-test-2,"",user_created_fork:fork-cruel-test-2,2012-10-23 23:49:23.867582
 
1762,3,demo,337,why-testing,"",started_following_repo,2012-10-23 23:53:03.878941
 
1763,3,demo,337,why-testing,"",user_created_repo,2012-10-23 23:53:03.893321
 
1764,3,demo,338,temp11,"",started_following_repo,2012-10-24 10:43:46.379754
 
1765,3,demo,338,temp11,"",user_created_repo,2012-10-24 10:43:46.391127
 
1766,3,demo,339,test123,"",started_following_repo,2012-10-24 14:10:41.927001
 
1767,3,demo,339,test123,"",user_created_repo,2012-10-24 14:10:41.936239
 
1768,3,demo,339,releases/test123,"",user_updated_repo,2012-10-24 14:10:57.739304
 
1769,2,admin,339,releases/test123,146.48.87.66,push:87e9dcd8d3e375c218fdb747da52dc1551a93dc7,2012-10-24 14:12:02.63513
 
1770,292,Airframe,340,Testtestac,"",started_following_repo,2012-10-24 17:09:26.37836
 
1771,292,Airframe,340,Testtestac,"",user_created_repo,2012-10-24 17:09:26.387917
 
1772,293,imbehind,190,xxx,"",user_commented_revision:008d9c6f7c92636aaeda6f182bffecac6a464dd3,2012-10-24 20:38:59.535098
 
1773,3,demo,341,MicTest,"",started_following_repo,2012-10-24 20:59:34.176289
 
1774,3,demo,341,MicTest,"",user_created_repo,2012-10-24 20:59:34.186308
 
1775,3,demo,341,MicTest,"",push_local:327bf84d0d633a66f0c7ff4533de2415c9ff5a88,2012-10-24 21:00:28.379598
 
1776,295,mhassanzadeh,342,ttest,"",started_following_repo,2012-10-25 10:50:06.001658
 
1777,295,mhassanzadeh,342,ttest,"",user_created_repo,2012-10-25 10:50:06.011192
 
1807,306,foobar,,foobar,"",user_forked_repo:fork-foobar,2012-10-30 05:39:47.942232
 
1809,306,foobar,,foobar,24.67.50.255,push:0e3a7dcdd90bdb2a1f6dbe85e7b574e4f6304207,2012-10-30 05:50:43.083487
 
1781,3,demo,,fork-rawr,"",user_deleted_repo,2012-10-25 15:14:56.164313
 
1782,3,demo,343,test_git_repo,"",started_following_repo,2012-10-26 13:31:41.642802
 
1783,3,demo,343,test_git_repo,"",user_created_repo,2012-10-26 13:31:41.653854
 
1784,2,admin,99,another-fork-to-check-code-review,"",user_commented_revision:d5422faf648cc589425cd3b0dbf1f6dbf93036a0,2012-10-27 15:38:54.924143
 
1785,3,demo,344,sadfasdf,"",started_following_repo,2012-10-28 11:41:22.491275
 
1786,3,demo,344,sadfasdf,"",user_created_repo,2012-10-28 11:41:22.503388
 
1787,3,demo,345,fork-dvorak,"",started_following_repo,2012-10-29 09:43:08.276759
 
1788,3,demo,129,dvorak,"",user_forked_repo:fork-dvorak,2012-10-29 09:43:08.321733
 
1789,3,demo,345,fork-dvorak,"",user_created_fork:fork-dvorak,2012-10-29 09:43:08.329436
 
1790,3,demo,346,my-test-repo,"",started_following_repo,2012-10-29 09:45:29.684135
 
1791,3,demo,346,my-test-repo,"",user_created_repo,2012-10-29 09:45:29.694219
 
1792,3,demo,347,fork-aaa-project111,"",started_following_repo,2012-10-29 09:50:58.797507
 
1793,3,demo,68,aaa-project,"",user_forked_repo:fork-aaa-project111,2012-10-29 09:50:58.809098
 
1794,3,demo,347,fork-aaa-project111,"",user_created_fork:fork-aaa-project111,2012-10-29 09:50:58.816283
 
1795,2,admin,348,group/I-am-a-very-long-repository-name,"",started_following_repo,2012-10-29 15:51:18.519038
 
1796,2,admin,348,group/I-am-a-very-long-repository-name,"",admin_created_repo,2012-10-29 15:51:18.528651
 
1797,3,demo,34,a-very-long-usper-loooong-name-that-is-to-long,"",user_updated_repo,2012-10-29 17:32:06.365281
 
1798,3,demo,350,misisko-repo,"",started_following_repo,2012-10-29 20:19:56.180975
 
1799,3,demo,350,misisko-repo,"",user_created_repo,2012-10-29 20:19:56.190032
 
1800,3,demo,350,misisko-repo,195.168.239.153,push:34bd2e2b6cf2abdb0e026182b9342f95d8f6f440,2012-10-29 20:21:33.067763
 
1801,3,demo,350,misisko-repo,"",user_commented_revision:34bd2e2b6cf2abdb0e026182b9342f95d8f6f440,2012-10-29 20:23:32.816207
 
1802,2,admin,350,misisko-repo,"",push_local:89a041cce422153ba0716a2beb2b5711ee4f4280,2012-10-29 21:23:56.649892
 
1806,306,foobar,,fork-foobar,"",started_following_repo,2012-10-30 05:39:47.86196
 
1808,306,foobar,,fork-foobar,"",user_created_fork:fork-foobar,2012-10-30 05:39:47.956653
 
1811,306,foobar,,fork-foobar,"",user_deleted_repo,2012-10-30 05:56:06.059093
 
1803,306,foobar,,foobar,"",started_following_repo,2012-10-30 05:36:49.915108
 
1804,306,foobar,,foobar,"",user_created_repo,2012-10-30 05:36:49.929097
 
1805,306,foobar,,foobar,24.67.50.255,push:3b0ad3c9b89dec8bcac5ec8bc9846f307dd5f291,2012-10-30 05:38:30.960676
 
1810,306,foobar,,foobar,24.67.50.255,push:5066e223f81d3983b9bc4d959c509bbce326474c,2012-10-30 05:53:14.168715
 
1812,306,foobar,,foobar,"",user_deleted_repo,2012-10-30 05:56:35.211293
 
1813,3,demo,354,DibaDaba,"",started_following_repo,2012-10-30 10:55:03.068862
 
1814,3,demo,354,DibaDaba,"",user_created_repo,2012-10-30 10:55:03.088077
 
1815,3,demo,354,DibaDaba,"",push_local:cf41636a602d2a69128cc58fdf441857afb83827,2012-10-30 10:56:16.451197
 
1816,3,demo,354,DibaDaba,"",user_updated_repo,2012-10-30 11:05:50.836108
 
1817,3,demo,354,DibaDaba,"",push_local:15453a2035510dcefbc8b7f3e45b916003ae4a0f,2012-10-30 11:18:42.79919
 
1818,3,demo,355,Test-RohdeCode,"",started_following_repo,2012-10-30 21:07:27.016976
 
1819,3,demo,355,Test-RohdeCode,"",user_created_repo,2012-10-30 21:07:27.040381
 
1820,3,demo,355,Test-RohdeCode,"",push_local:a2c7c944a5bae039181a84bec8bc89a9bb4ac68c,2012-10-31 10:52:21.038578
 
1821,3,demo,355,Test-RohdeCode,"",push_local:303c0051c7a04b5a4f77654d8cfeb09a61322e15,2012-10-31 16:32:16.917289
 
1822,3,demo,356,testdzone,"",started_following_repo,2012-11-02 15:51:18.982652
 
1823,3,demo,356,testdzone,"",user_created_repo,2012-11-02 15:51:18.994144
 
1824,3,demo,357,firs,"",started_following_repo,2012-11-04 00:37:08.689703
 
1825,3,demo,357,firs,"",user_created_repo,2012-11-04 00:37:08.700977
 
1826,316,theo,358,brightandshiny,"",started_following_repo,2012-11-04 14:17:04.270938
 
1827,316,theo,358,brightandshiny,"",user_created_repo,2012-11-04 14:17:04.286049
 
1828,316,theo,358,brightandshiny,87.165.13.203,"push:831fd48093c38f64bb541a4c12bec45d27c13f9d,0177264991ad113c2dbeed00716738931bc0222f",2012-11-04 14:21:33.759458
 
1829,3,demo,45,group/fork-bootstrap,"",started_following_repo,2012-11-04 18:25:45.459487
 
1830,3,demo,45,group/fork-bootstrap,"",stopped_following_repo,2012-11-04 18:25:48.166886
 
1831,3,demo,45,group/fork-bootstrap,"",started_following_repo,2012-11-04 18:25:51.905362
 
1832,3,demo,45,group/fork-bootstrap,"",stopped_following_repo,2012-11-04 18:25:54.406445
 
1833,3,demo,227,fork-aaa-project-2,"",user_commented_revision:bc829c34912709c7d129e5dae9a3e5ff46e77fc7,2012-11-05 18:37:48.772572
 
1834,3,demo,227,fork-aaa-project-2,"",user_commented_revision:bc829c34912709c7d129e5dae9a3e5ff46e77fc7,2012-11-05 18:38:42.521316
 
1835,3,demo,359,MyUberRepo,"",started_following_repo,2012-11-05 22:04:02.939575
 
1836,3,demo,359,MyUberRepo,"",user_created_repo,2012-11-05 22:04:02.95609
 
1837,3,demo,359,MyUberRepo,"",push_local:a67ddbdca6d533e0cd43175068cc12f526b2f95f,2012-11-05 22:07:40.846096
 
1838,3,demo,360,marese_test,"",started_following_repo,2012-11-05 22:10:35.535744
 
1839,3,demo,360,marese_test,"",user_created_repo,2012-11-05 22:10:35.547878
 
1840,3,demo,361,test-23,"",started_following_repo,2012-11-06 04:01:46.385817
 
1841,3,demo,361,test-23,"",user_created_repo,2012-11-06 04:01:46.410423
 
1842,3,demo,361,test-23,"",push_local:7d8547b154a01c7055a09292e871e80f3c6b574b,2012-11-06 04:03:03.320736
 
1843,3,demo,361,test-23,"",user_commented_revision:7d8547b154a01c7055a09292e871e80f3c6b574b,2012-11-06 04:07:20.51212
 
1844,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:12:51.753053
 
1845,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:13:30.727101
 
1846,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:15:29.032768
 
1847,3,demo,68,aaa-project,"",user_commented_revision:0c33fa58efc5a5541d8d3f1a3d3b77367e3d94f5,2012-11-06 04:17:16.091927
 
1848,3,demo,249,abcdefg,"",user_commented_revision:5d80e28538141e322b317168e2367fb03178d58c,2012-11-06 06:10:53.372505
 
1849,3,demo,362,Marcin,"",started_following_repo,2012-11-06 10:45:38.505485
 
1850,3,demo,362,Marcin,"",user_created_repo,2012-11-06 10:45:38.518969
 
1851,3,demo,362,hidden/Marcin,"",user_updated_repo,2012-11-06 10:45:46.281581
 
1849,3,demo,362,Username,"",started_following_repo,2012-11-06 10:45:38.505485
 
1850,3,demo,362,Username,"",user_created_repo,2012-11-06 10:45:38.518969
 
1851,3,demo,362,hidden/Username,"",user_updated_repo,2012-11-06 10:45:46.281581
 
1852,3,demo,177,blah,62.200.22.2,push:b369fb18c8d61fe0d3b14c417466680230cabe46,2012-11-06 10:47:55.655029
 
1853,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:d5422faf648cc589425cd3b0dbf1f6dbf93036a0,2012-11-06 13:12:05.517155
 
1854,3,demo,38,code-review-test,"",user_commented_revision:6d7db5794e8cad7da042b6ae6238116c6e59a4d2,2012-11-06 16:12:59.38977
 
1855,3,demo,363,utf8-test,"",started_following_repo,2012-11-06 16:52:34.020264
 
1856,3,demo,363,utf8-test,"",user_created_repo,2012-11-06 16:52:34.036171
 
1857,3,demo,363,utf8-test,178.6.80.143,push:e9e69068b8277c8e857984fb44e54d4e057710bd,2012-11-06 16:54:30.357337
 
1858,320,blahblub,364,abc_test,"",started_following_repo,2012-11-06 22:06:45.476725
 
1859,320,blahblub,364,abc_test,"",user_created_repo,2012-11-06 22:06:45.488546
 
1860,320,blahblub,364,abc_test,81.190.220.154,"push:e05d82d631f4a61972c2c76eb5b473c55ee6133f,dccf54ebc0638183872cfd9d35f0f49aa44ad4bc",2012-11-06 22:07:39.657839
 
1861,320,blahblub,364,abc_test,81.190.220.154,push:8ef2f06b68eb25067c4ca3a9c73c89bb681f031b,2012-11-06 22:09:18.572762
 
1862,3,demo,365,testing_do,"",started_following_repo,2012-11-07 00:19:56.15291
 
1863,3,demo,365,testing_do,"",user_created_repo,2012-11-07 00:19:56.165626
 
1864,3,demo,365,testing_do,"",user_updated_repo,2012-11-07 00:21:24.482025
 
1865,3,demo,321,Mecurial-Test,"",user_commented_revision:041e48fd7acad4c27385c27a4789d3d5cc7ccbea,2012-11-07 01:16:37.129361
 
1866,321,ifco89,284,fork-django,"",user_commented_revision:129f1ac8484d63c2e61a44fb2a18dd17246c1c4d,2012-11-07 01:26:40.887017
 
1867,3,demo,284,fork-django,"",user_commented_revision:c5d6f6d6829e730bdddf63c1252304f0c49a9053,2012-11-07 01:27:33.995596
 
1868,3,demo,68,aaa-project,"",push_local:5ecd97456233badb3680873dc5b7d4bfb52ede79,2012-11-07 06:43:22.378406
 
1869,3,demo,319,fork1-code-review-test,"",user_commented_revision:e216971a777e3132487405dd2c8e093f66788765,2012-11-07 08:44:19.419642
 
1870,3,demo,366,test123123,"",started_following_repo,2012-11-07 14:16:13.029479
 
1871,3,demo,366,test123123,"",user_created_repo,2012-11-07 14:16:13.04346
 
1872,323,didi,364,abc_test,"",user_commented_revision:8ef2f06b68eb25067c4ca3a9c73c89bb681f031b,2012-11-07 15:45:10.378961
 
1873,324,tutty,367,fork-bootstrap,"",started_following_repo,2012-11-07 16:00:47.116111
 
1874,324,tutty,36,bootstrap,"",user_forked_repo:fork-bootstrap,2012-11-07 16:00:48.378051
 
1875,324,tutty,367,fork-bootstrap,"",user_created_fork:fork-bootstrap,2012-11-07 16:00:48.393002
 
1876,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:07.016059
 
1877,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:34.311311
 
1878,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:38.071834
 
1879,3,demo,368,Repotest,"",started_following_repo,2012-11-09 10:40:40.804085
 
1880,3,demo,368,Repotest,"",user_created_repo,2012-11-09 10:40:40.813114
 
1881,3,demo,369,fork-HelloWorld,"",started_following_repo,2012-11-10 19:57:46.238558
 
1882,3,demo,280,HelloWorld,"",user_forked_repo:fork-HelloWorld,2012-11-10 19:57:46.306001
 
1883,3,demo,369,fork-HelloWorld,"",user_created_fork:fork-HelloWorld,2012-11-10 19:57:46.320675
 
1884,3,demo,370,fork-django-more,"",started_following_repo,2012-11-11 01:18:51.709667
 
1885,3,demo,110,django,"",user_forked_repo:fork-django-more,2012-11-11 01:18:58.716934
 
1886,3,demo,370,fork-django-more,"",user_created_fork:fork-django-more,2012-11-11 01:18:58.731127
 
1887,3,demo,371,hgh,"",started_following_repo,2012-11-12 06:07:06.509759
 
1888,3,demo,371,hgh,"",user_created_repo,2012-11-12 06:07:06.525795
 
1889,3,demo,371,hgh,"",push_local:b43f437976ff61a8fa37709f105091ffd00bbd2c,2012-11-12 06:08:34.009007
 
1890,3,demo,371,hgh,"",push_local:93c5a23d3cbde4c6dbd21d722eb1beff90a1d67c,2012-11-12 06:09:44.749181
 
1891,3,demo,371,hgh,"",push_local:d3e97cf131b0feaa586272d88e71e97a65564806,2012-11-12 06:13:04.240444
 
1892,336,demodemo,373,test-for,"",started_following_repo,2012-11-12 18:29:12.02868
 
1893,336,demodemo,373,test-for,"",user_created_repo,2012-11-12 18:29:12.041121
 
1894,336,demodemo,374,fork-test-for,"",started_following_repo,2012-11-12 18:29:25.869055
 
1895,336,demodemo,373,test-for,"",user_forked_repo:fork-test-for,2012-11-12 18:29:25.922561
 
1896,336,demodemo,374,fork-test-for,"",user_created_fork:fork-test-for,2012-11-12 18:29:25.936869
 
1897,336,demodemo,374,fork-test-for,101.140.96.42,push:a7495b86f0bfc299740cdbfbbca1b9d5ca1f62de,2012-11-12 18:31:51.107494
 
1898,2,admin,376,linux-kernel,"",started_following_repo,2012-11-13 22:14:17.466937
 
1899,342,gauravve,377,myrep,"",started_following_repo,2012-11-14 12:12:46.280751
 
1900,342,gauravve,377,myrep,"",user_created_repo,2012-11-14 12:12:46.296606
 
1901,343,yasser.a,125,test-to-end-all-tests,"",user_commented_revision:fb4eb182028a674ae250430c04305e9b3fbe508e,2012-11-14 16:17:28.970017
 
772,2,admin,,cruel-test,"",started_following_repo,2012-08-28 10:23:44.560576
 
1902,254,yujiro,378,arai-new-repo,"",started_following_repo,2012-11-15 02:39:50.649631
 
1903,254,yujiro,378,arai-new-repo,"",user_created_repo,2012-11-15 02:39:50.667177
 
1904,254,yujiro,378,arai-new-repo,182.171.245.184,push:6beb6da3b0e27eb1401090e122f7675d56e07fc1,2012-11-15 02:45:25.207338
 
1905,3,demo,381,test01012145,"",started_following_repo,2012-11-16 13:02:38.980282
 
1906,3,demo,381,test01012145,"",user_created_repo,2012-11-16 13:02:38.989119
 
1907,3,demo,381,test01012145,"",push_local:225a9898957de5f78ac2aa8e4d46a80eed3ee6d5,2012-11-16 13:04:53.384353
 
1908,3,demo,382,aaa-repo-fork,"",started_following_repo,2012-11-17 12:58:42.919876
 
1909,3,demo,67,group/aaa-repo,"",user_forked_repo:aaa-repo-fork,2012-11-17 12:58:42.981111
 
1910,3,demo,382,aaa-repo-fork,"",user_created_fork:aaa-repo-fork,2012-11-17 12:58:42.992071
 
1911,3,demo,382,aaa-repo-fork,"",user_updated_repo,2012-11-17 12:59:32.084662
 
1912,2,admin,,"","",admin_updated_user:kiall,2012-11-17 21:52:15.776253
 
1913,3,demo,268,foofootest,"",user_commented_revision:4bc0645c20363db85c9a5050a85c2c05a1991765,2012-11-20 12:19:44.614295
 
1914,3,demo,383,aaa-test-repo,"",started_following_repo,2012-11-20 12:27:16.120325
 
1915,3,demo,383,aaa-test-repo,"",user_created_repo,2012-11-20 12:27:16.130553
 
1916,3,demo,383,aaa-test-repo,"",push_local:fb2f5ae9a806abe4cb03fe7b11d4ad21a836f7b6,2012-11-20 12:27:41.98988
 
1917,3,demo,384,aaa-test-repo2,"",started_following_repo,2012-11-20 12:31:43.009679
 
1918,3,demo,384,aaa-test-repo2,"",user_created_repo,2012-11-20 12:31:43.019015
 
1919,3,demo,384,aaa-test-repo2,"",push_local:83a7be8f9d651912ee343067bd392678fbebe3e8,2012-11-20 12:32:34.336741
 
1920,3,demo,385,aaa-test-repo2-fork,"",started_following_repo,2012-11-20 12:58:25.715274
 
1921,3,demo,384,aaa-test-repo2,"",user_forked_repo:aaa-test-repo2-fork,2012-11-20 12:58:25.729374
 
1922,3,demo,385,aaa-test-repo2-fork,"",user_created_fork:aaa-test-repo2-fork,2012-11-20 12:58:25.737204
 
1923,3,demo,385,aaa-test-repo2-fork,"",push_local:af794c0c7cf39be92c6b59bb8b7871d313b1d248,2012-11-20 12:58:59.668087
 
1924,3,demo,384,aaa-test-repo2,"",user_commented_pull_request:74,2012-11-20 13:25:32.773937
 
1925,3,demo,384,aaa-test-repo2,"",user_commented_pull_request:74,2012-11-20 13:33:39.823854
 
1926,3,demo,384,aaa-test-repo2,"",user_closed_pull_request:74,2012-11-20 13:33:39.828583
 
1927,3,demo,385,aaa-test-repo2-fork,"",push_local:67120aaf6ddd81cd2aabde6adc22e7ca90d15d12,2012-11-20 18:41:48.280705
 
1928,3,demo,384,aaa-test-repo2,81.178.131.20,"push:af794c0c7cf39be92c6b59bb8b7871d313b1d248,67120aaf6ddd81cd2aabde6adc22e7ca90d15d12",2012-11-20 20:01:34.101349
 
1929,3,demo,385,aaa-test-repo2-fork,"",push_local:5ece07f12c61645049e7d2d3d33797942b16cba5,2012-11-20 20:10:49.816215
 
1930,3,demo,385,aaa-test-repo2-fork,"",push_local:fd8a6936e27e3c31684155790f78effb33e6cc5c,2012-11-20 20:24:06.305599
 
1931,3,demo,384,aaa-test-repo2,81.178.131.20,push:5ece07f12c61645049e7d2d3d33797942b16cba5,2012-11-20 20:49:00.180794
 
1932,3,demo,384,aaa-test-repo2,81.178.131.20,push:fd8a6936e27e3c31684155790f78effb33e6cc5c,2012-11-20 21:28:16.281326
 
1933,3,demo,384,aaa-test-repo2,"",user_commented_pull_request:75,2012-11-21 01:38:22.165876
 
1934,3,demo,384,aaa-test-repo2,"",user_closed_pull_request:75,2012-11-21 01:38:22.175975
 
1935,3,demo,385,aaa-test-repo2-fork,"",push_local:586be5737b6b185d37ab249e028470417698b44d,2012-11-21 01:42:21.540009
 
1936,3,demo,385,aaa-test-repo2-fork,"",push_local:b836457cf5b358555c26193e67770d56e2345f4c,2012-11-21 01:50:03.178941
 
1937,3,demo,384,aaa-test-repo2,81.178.131.20,"push:586be5737b6b185d37ab249e028470417698b44d,b836457cf5b358555c26193e67770d56e2345f4c",2012-11-21 02:01:01.40853
 
1938,3,demo,34,a-very-long-usper-loooong-name-that-is-to-long,"",push_local:ff712eae5a99bede2bf710d86f86b0b534fbf089,2012-11-21 02:09:50.150788
 
1939,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-21 08:36:43.456525
 
1940,3,demo,384,aaa-test-repo2,"",user_commented_revision:b836457cf5b358555c26193e67770d56e2345f4c,2012-11-21 12:03:06.317362
 
1941,3,demo,385,aaa-test-repo2-fork,"",push_local:d38a759ba1f79489db44854c9ddf06986f7e2b72,2012-11-21 12:17:04.303085
 
1942,3,demo,385,aaa-test-repo2-fork,"",user_commented_revision:d38a759ba1f79489db44854c9ddf06986f7e2b72,2012-11-21 12:17:47.536221
 
1943,3,demo,34,a-very-long-usper-loooong-name-that-is-to-long,"",user_commented_revision:f437e821a406d35b1e231c5a81e82853f8ed6436,2012-11-21 14:27:15.134067
 
1944,3,demo,386,I-am-a-very-long-repository-name-fork,"",started_following_repo,2012-11-21 15:31:17.25368
 
1945,3,demo,348,group/I-am-a-very-long-repository-name,"",user_forked_repo:I-am-a-very-long-repository-name-fork,2012-11-21 15:31:17.271794
 
1946,3,demo,386,I-am-a-very-long-repository-name-fork,"",user_created_fork:I-am-a-very-long-repository-name-fork,2012-11-21 15:31:17.283533
kallithea/tests/functional/test_login.py
➞
Show inline comments
 
@@ -115,235 +115,235 @@ class TestLoginController(TestController
 

			




	
	
	
		
 
        msg = validators.ValidUsername()._messages['username_exists']

			




	
	
	
		
 
        msg = h.html_escape(msg % {'username': uname})

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_err_same_email(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': 'test_admin_0',

			




	
	
	
		
 
                                             'password': 'test12',

			




	
	
	
		
 
                                             'password_confirmation': 'test12',

			




	
	
	
		
 
                                             'email': 'test_admin@mail.com',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 

			




	
	
	
		
 
        msg = validators.UniqSystemEmail()()._messages['email_taken']

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_err_same_email_case_sensitive(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': 'test_admin_1',

			




	
	
	
		
 
                                             'password': 'test12',

			




	
	
	
		
 
                                             'password_confirmation': 'test12',

			




	
	
	
		
 
                                             'email': 'TesT_Admin@mail.COM',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 
        msg = validators.UniqSystemEmail()()._messages['email_taken']

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_err_wrong_data(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': 'xs',

			




	
	
	
		
 
                                             'password': 'test',

			




	
	
	
		
 
                                             'password_confirmation': 'test',

			




	
	
	
		
 
                                             'email': 'goodmailm',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 
        self.assertEqual(response.status, '200 OK')

			




	
	
	
		
 
        response.mustcontain('An email address must contain a single @')

			




	
	
	
		
 
        response.mustcontain('Enter a value 6 characters long or more')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_err_username(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': 'error user',

			




	
	
	
		
 
                                             'password': 'test12',

			




	
	
	
		
 
                                             'password_confirmation': 'test12',

			




	
	
	
		
 
                                             'email': 'goodmailm',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 

			




	
	
	
		
 
        response.mustcontain('An email address must contain a single @')

			




	
	
	
		
 
        response.mustcontain('Username may only contain '

			




	
	
	
		
 
                'alphanumeric characters underscores, '

			




	
	
	
		
 
                'periods or dashes and must begin with '

			




	
	
	
		
 
                'alphanumeric character')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_err_case_sensitive(self):

			




	
	
	
		
 
        usr = 'Test_Admin'

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': usr,

			




	
	
	
		
 
                                             'password': 'test12',

			




	
	
	
		
 
                                             'password_confirmation': 'test12',

			




	
	
	
		
 
                                             'email': 'goodmailm',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 

			




	
	
	
		
 
        response.mustcontain('An email address must contain a single @')

			




	
	
	
		
 
        msg = validators.ValidUsername()._messages['username_exists']

			




	
	
	
		
 
        msg = h.html_escape(msg % {'username': usr})

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_special_chars(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                        {'username': 'xxxaxn',

			




	
	
	
		
 
                                         'password': 'ąćźżąśśśś',

			




	
	
	
		
 
                                         'password_confirmation': 'ąćźżąśśśś',

			




	
	
	
		
 
                                         'email': 'goodmailm@test.plx',

			




	
	
	
		
 
                                         'firstname': 'test',

			




	
	
	
		
 
                                         'lastname': 'test'})

			




	
	
	
		
 

			




	
	
	
		
 
        msg = validators.ValidPassword()._messages['invalid_password']

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_password_mismatch(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': 'xs',

			




	
	
	
		
 
                                             'password': '123qwe',

			




	
	
	
		
 
                                             'password_confirmation': 'qwe123',

			




	
	
	
		
 
                                             'email': 'goodmailm@test.plxa',

			




	
	
	
		
 
                                             'firstname': 'test',

			




	
	
	
		
 
                                             'lastname': 'test'})

			




	
	
	
		
 
        msg = validators.ValidPasswordsMatch()._messages['password_mismatch']

			




	
	
	
		
 
        response.mustcontain(msg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_register_ok(self):

			




	
	
	
		
 
        username = 'test_regular4'

			




	
	
	
		
 
        password = 'qweqwe'

			




	
	
	
		
 
        email = 'marcin@test.com'

			




	
	
	
		
 
        email = 'username@test.com'

			




	
	
	
		
 
        name = 'testname'

			




	
	
	
		
 
        lastname = 'testlastname'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='register'),

			




	
	
	
		
 
                                            {'username': username,

			




	
	
	
		
 
                                             'password': password,

			




	
	
	
		
 
                                             'password_confirmation': password,

			




	
	
	
		
 
                                             'email': email,

			




	
	
	
		
 
                                             'firstname': name,

			




	
	
	
		
 
                                             'lastname': lastname,

			




	
	
	
		
 
                                             'admin': True})  # This should be overriden

			




	
	
	
		
 
        self.assertEqual(response.status, '302 Found')

			




	
	
	
		
 
        self.checkSessionFlash(response, 'You have successfully registered into Kallithea')

			




	
	
	
		
 

			




	
	
	
		
 
        ret = Session().query(User).filter(User.username == 'test_regular4').one()

			




	
	
	
		
 
        self.assertEqual(ret.username, username)

			




	
	
	
		
 
        self.assertEqual(check_password(password, ret.password), True)

			




	
	
	
		
 
        self.assertEqual(ret.email, email)

			




	
	
	
		
 
        self.assertEqual(ret.name, name)

			




	
	
	
		
 
        self.assertEqual(ret.lastname, lastname)

			




	
	
	
		
 
        self.assertNotEqual(ret.api_key, None)

			




	
	
	
		
 
        self.assertEqual(ret.admin, False)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_forgot_password_wrong_mail(self):

			




	
	
	
		
 
        bad_email = 'marcin@wrongmail.org'

			




	
	
	
		
 
        bad_email = 'username@wrongmail.org'

			




	
	
	
		
 
        response = self.app.post(

			




	
	
	
		
 
                        url(controller='login', action='password_reset'),

			




	
	
	
		
 
                            {'email': bad_email, }

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 
        msg = validators.ValidSystemEmail()._messages['non_existing_email']

			




	
	
	
		
 
        msg = h.html_escape(msg % {'email': bad_email})

			




	
	
	
		
 
        response.mustcontain()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_forgot_password(self):

			




	
	
	
		
 
        response = self.app.get(url(controller='login',

			




	
	
	
		
 
                                    action='password_reset'))

			




	
	
	
		
 
        self.assertEqual(response.status, '200 OK')

			




	
	
	
		
 

			




	
	
	
		
 
        username = 'test_password_reset_1'

			




	
	
	
		
 
        password = 'qweqwe'

			




	
	
	
		
 
        email = 'marcin@python-works.com'

			




	
	
	
		
 
        email = 'username@python-works.com'

			




	
	
	
		
 
        name = 'passwd'

			




	
	
	
		
 
        lastname = 'reset'

			




	
	
	
		
 

			




	
	
	
		
 
        new = User()

			




	
	
	
		
 
        new.username = username

			




	
	
	
		
 
        new.password = password

			




	
	
	
		
 
        new.email = email

			




	
	
	
		
 
        new.name = name

			




	
	
	
		
 
        new.lastname = lastname

			




	
	
	
		
 
        new.api_key = generate_api_key(username)

			




	
	
	
		
 
        Session().add(new)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url(controller='login',

			




	
	
	
		
 
                                     action='password_reset'),

			




	
	
	
		
 
                                 {'email': email, })

			




	
	
	
		
 

			




	
	
	
		
 
        self.checkSessionFlash(response, 'Your password reset link was sent')

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 

			




	
	
	
		
 
        # BAD KEY

			




	
	
	
		
 

			




	
	
	
		
 
        key = "bad"

			




	
	
	
		
 
        response = self.app.get(url(controller='login',

			




	
	
	
		
 
                                    action='password_reset_confirmation',

			




	
	
	
		
 
                                    key=key))

			




	
	
	
		
 
        self.assertEqual(response.status, '302 Found')

			




	
	
	
		
 
        self.assertTrue(response.location.endswith(url('reset_password')))

			




	
	
	
		
 

			




	
	
	
		
 
        # GOOD KEY

			




	
	
	
		
 

			




	
	
	
		
 
        key = User.get_by_username(username).api_key

			




	
	
	
		
 
        response = self.app.get(url(controller='login',

			




	
	
	
		
 
                                    action='password_reset_confirmation',

			




	
	
	
		
 
                                    key=key))

			




	
	
	
		
 
        self.assertEqual(response.status, '302 Found')

			




	
	
	
		
 
        self.assertTrue(response.location.endswith(url('login_home')))

			




	
	
	
		
 

			




	
	
	
		
 
        self.checkSessionFlash(response,

			




	
	
	
		
 
                               ('Your password reset was successful, '

			




	
	
	
		
 
                                'new password has been sent to your email'))

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 

			




	
	
	
		
 
    def _get_api_whitelist(self, values=None):

			




	
	
	
		
 
        config = {'api_access_controllers_whitelist': values or []}

			




	
	
	
		
 
        return config

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        ('none', None),

			




	
	
	
		
 
        ('empty_string', ''),

			




	
	
	
		
 
        ('fake_number', '123456'),

			




	
	
	
		
 
        ('proper_api_key', None)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):

			




	
	
	
		
 
        whitelist = self._get_api_whitelist([])

			




	
	
	
		
 
        with mock.patch('kallithea.CONFIG', whitelist):

			




	
	
	
		
 
            self.assertEqual([],

			




	
	
	
		
 
                             whitelist['api_access_controllers_whitelist'])

			




	
	
	
		
 
            if test_name == 'proper_api_key':

			




	
	
	
		
 
                #use builtin if api_key is None

			




	
	
	
		
 
                api_key = User.get_first_admin().api_key

			




	
	
	
		
 

			




	
	
	
		
 
            with fixture.anon_access(False):

			




	
	
	
		
 
                self.app.get(url(controller='changeset',

			




	
	
	
		
 
                                 action='changeset_raw',

			




	
	
	
		
 
                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

			




	
	
	
		
 
                             status=302)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        ('none', None, 302),

			




	
	
	
		
 
        ('empty_string', '', 302),

			




	
	
	
		
 
        ('fake_number', '123456', 302),

			




	
	
	
		
 
        ('proper_api_key', None, 200)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):

			




	
	
	
		
 
        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

			




	
	
	
		
 
        with mock.patch('kallithea.CONFIG', whitelist):

			




	
	
	
		
 
            self.assertEqual(['ChangesetController:changeset_raw'],

			




	
	
	
		
 
                             whitelist['api_access_controllers_whitelist'])

			




	
	
	
		
 
            if test_name == 'proper_api_key':

			




	
	
	
		
 
                api_key = User.get_first_admin().api_key

			




	
	
	
		
 

			




	
	
	
		
 
            with fixture.anon_access(False):

			




	
	
	
		
 
                self.app.get(url(controller='changeset',

			




	
	
	
		
 
                                 action='changeset_raw',

			




	
	
	
		
 
                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

			




	
	
	
		
 
                             status=code)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_access_page_via_extra_api_key(self):

			




	
	
	
		
 
        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

			




	
	
	
		
 
        with mock.patch('kallithea.CONFIG', whitelist):

			




	
	
	
		
 
            self.assertEqual(['ChangesetController:changeset_raw'],

			




	
	
	
		
 
                             whitelist['api_access_controllers_whitelist'])

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/other/manual_test_vcs_operations.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -18,193 +18,193 @@ kallithea.tests.test_scm_operations

			




	
	
	
		
 
Test suite for making push/pull operations.

			




	
	
	
		
 

			




	
	
	
		
 
Run it in two terminals::

			




	
	
	
		
 
 paster serve test.ini

			




	
	
	
		
 
 KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 nosetests kallithea/tests/other/manual_test_vcs_operations.py

			




	
	
	
		
 

			




	
	
	
		
 
You must have git > 1.8.1 for tests to work fine

			




	
	
	
		
 

			




	
	
	
		
 
This file was forked by the Kallithea project in July 2014.

			




	
	
	
		
 
Original author and date, and relevant copyright and licensing information is below:

			




	
	
	
		
 
:created_on: Dec 30, 2010

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
import time

			




	
	
	
		
 
from os.path import join as jn

			




	
	
	
		
 

			




	
	
	
		
 
from tempfile import _RandomNameSequence

			




	
	
	
		
 
from subprocess import Popen, PIPE

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 
from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.repo import RepoModel

			




	
	
	
		
 
from kallithea.model.user import UserModel

			




	
	
	
		
 

			




	
	
	
		
 
DEBUG = True

			




	
	
	
		
 
HOST = '127.0.0.1:5000'  # test host

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class Command(object):

			




	
	
	
		
 

			




	
	
	
		
 
    def __init__(self, cwd):

			




	
	
	
		
 
        self.cwd = cwd

			




	
	
	
		
 

			




	
	
	
		
 
    def execute(self, cmd, *args):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Runs command on the system with given ``args``.

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        command = cmd + ' ' + ' '.join(args)

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print '*** CMD %s ***' % command

			




	
	
	
		
 
        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)

			




	
	
	
		
 
        stdout, stderr = p.communicate()

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print 'stdout:', repr(stdout)

			




	
	
	
		
 
            print 'stderr:', repr(stderr)

			




	
	
	
		
 
        return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_tmp_dir():

			




	
	
	
		
 
    return tempfile.mkdtemp(prefix='rc_integration_test')

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _construct_url(repo, dest=None, **kwargs):

			




	
	
	
		
 
    if dest is None:

			




	
	
	
		
 
        #make temp clone

			




	
	
	
		
 
        dest = _get_tmp_dir()

			




	
	
	
		
 
    params = {

			




	
	
	
		
 
        'user': TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
        'passwd': TEST_USER_ADMIN_PASS,

			




	
	
	
		
 
        'host': HOST,

			




	
	
	
		
 
        'cloned_repo': repo,

			




	
	
	
		
 
        'dest': dest

			




	
	
	
		
 
    }

			




	
	
	
		
 
    params.update(**kwargs)

			




	
	
	
		
 
    if params['user'] and params['passwd']:

			




	
	
	
		
 
        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params

			




	
	
	
		
 
    return _url

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _add_files_and_push(vcs, DEST, **kwargs):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Generate some files, add it to DEST repo and push back

			




	
	
	
		
 
    vcs is git or hg and defines what VCS we want to make those files for

			




	
	
	
		
 

			




	
	
	
		
 
    :param vcs:

			




	
	
	
		
 
    :param DEST:

			




	
	
	
		
 
    """

			




	
	
	
		
 
    # commit some stuff into this repo

			




	
	
	
		
 
    cwd = path = jn(DEST)

			




	
	
	
		
 
    #added_file = jn(path, '%ssetupążźć.py' % _RandomNameSequence().next())

			




	
	
	
		
 
    added_file = jn(path, '%ssetup.py' % _RandomNameSequence().next())

			




	
	
	
		
 
    Command(cwd).execute('touch %s' % added_file)

			




	
	
	
		
 
    Command(cwd).execute('%s add %s' % (vcs, added_file))

			




	
	
	
		
 

			




	
	
	
		
 
    for i in xrange(kwargs.get('files_no', 3)):

			




	
	
	
		
 
        cmd = """echo 'added_line%s' >> %s""" % (i, added_file)

			




	
	
	
		
 
        Command(cwd).execute(cmd)

			




	
	
	
		
 
        author_str = 'Marcin Kuźminski <me@email.com>'

			




	
	
	
		
 
        author_str = 'User ǝɯɐᴎ <me@email.com>'

			




	
	
	
		
 
        if vcs == 'hg':

			




	
	
	
		
 
            cmd = """hg commit -m 'commited new %s' -u '%s' %s """ % (

			




	
	
	
		
 
                i, author_str, added_file

			




	
	
	
		
 
            )

			




	
	
	
		
 
        elif vcs == 'git':

			




	
	
	
		
 
            cmd = """EMAIL="me@email.com" git commit -m 'commited new %s' --author '%s' %s """ % (

			




	
	
	
		
 
                i, author_str, added_file

			




	
	
	
		
 
            )

			




	
	
	
		
 
        Command(cwd).execute(cmd)

			




	
	
	
		
 

			




	
	
	
		
 
    # PUSH it back

			




	
	
	
		
 
    _REPO = None

			




	
	
	
		
 
    if vcs == 'hg':

			




	
	
	
		
 
        _REPO = HG_REPO

			




	
	
	
		
 
    elif vcs == 'git':

			




	
	
	
		
 
        _REPO = GIT_REPO

			




	
	
	
		
 

			




	
	
	
		
 
    kwargs['dest'] = ''

			




	
	
	
		
 
    clone_url = _construct_url(_REPO, **kwargs)

			




	
	
	
		
 
    if 'clone_url' in kwargs:

			




	
	
	
		
 
        clone_url = kwargs['clone_url']

			




	
	
	
		
 
    stdout = stderr = None

			




	
	
	
		
 
    if vcs == 'hg':

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)

			




	
	
	
		
 
    elif vcs == 'git':

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")

			




	
	
	
		
 

			




	
	
	
		
 
    return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def set_anonymous_access(enable=True):

			




	
	
	
		
 
    user = User.get_by_username(User.DEFAULT_USER)

			




	
	
	
		
 
    user.active = enable

			




	
	
	
		
 
    Session().add(user)

			




	
	
	
		
 
    Session().commit()

			




	
	
	
		
 
    print '\tanonymous access is now:', enable

			




	
	
	
		
 
    if enable != User.get_by_username(User.DEFAULT_USER).active:

			




	
	
	
		
 
        raise Exception('Cannot set anonymous access')

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
#==============================================================================

			




	
	
	
		
 
# TESTS

			




	
	
	
		
 
#==============================================================================

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _check_proper_git_push(stdout, stderr):

			




	
	
	
		
 
    #WTF Git stderr is output ?!

			




	
	
	
		
 
    assert 'fatal' not in stderr

			




	
	
	
		
 
    assert 'rejected' not in stderr

			




	
	
	
		
 
    assert 'Pushing to' in stderr

			




	
	
	
		
 
    assert 'master -> master' in stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestVCSOperations(BaseTestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def setup_class(cls):

			




	
	
	
		
 
        #DISABLE ANONYMOUS ACCESS

			




	
	
	
		
 
        set_anonymous_access(False)

			




	
	
	
		
 

			




	
	
	
		
 
    def setUp(self):

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
	
		
 
        Repository.unlock(r)

			




	
	
	
		
 
        r.enable_locking = False

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
	
		
 
        Repository.unlock(r)

			




	
	
	
		
 
        r.enable_locking = False

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_by_admin(self):

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'requesting all changes' in stdout

			




	
	
	
		
 
        assert 'adding changesets' in stdout

			




	
	
	
		
 
        assert 'adding manifests' in stdout

			




	
	
	
		
 
        assert 'adding file changes' in stdout

			




	
	
	
		
 

			




	
	
	
		
 
        assert stderr == ''

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_repo_by_admin(self):

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'Cloning into' in stdout + stderr

			




	
	
	
		
 
        assert stderr == '' or stdout == ''

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_wrong_credentials_hg(self):

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, passwd='bad!')

			




	
	
	
		
 
        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

			




	
	
	
		
 
        assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_libs.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 
# This program is free software: you can redistribute it and/or modify

			




	
	
	
		
 
# it under the terms of the GNU General Public License as published by

			




	
	
	
		
 
# the Free Software Foundation, either version 3 of the License, or

			




	
	
	
		
 
# (at your option) any later version.

			




	
	
	
		
 
#

			




	
	
	
		
 
# This program is distributed in the hope that it will be useful,

			




	
	
	
		
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of

			




	
	
	
		
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

			




	
	
	
		
 
# GNU General Public License for more details.

			




	
	
	
		
 
#

			




	
	
	
		
 
# You should have received a copy of the GNU General Public License

			




	
	
	
		
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

			




	
	
	
		
 
"""

			




	
	
	
		
 
kallithea.tests.test_libs

			




	
	
	
		
 
~~~~~~~~~~~~~~~~~~~~~~~~~

			




	
	
	
		
 

			




	
	
	
		
 
Package for testing various lib/helper functions in kallithea

			




	
	
	
		
 

			




	
	
	
		
 
This file was forked by the Kallithea project in July 2014.

			




	
	
	
		
 
Original author and date, and relevant copyright and licensing information is below:

			




	
	
	
		
 
:created_on: Jun 9, 2011

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
from __future__ import with_statement

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import hashlib

			




	
	
	
		
 
import mock

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 
from kallithea.lib.utils2 import AttributeDict

			




	
	
	
		
 
from kallithea.model.db import Repository

			




	
	
	
		
 

			




	
	
	
		
 
proto = 'http'

			




	
	
	
		
 
TEST_URLS = [

			




	
	
	
		
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

			




	
	
	
		
 
     '%s://127.0.0.1:8080' % proto),

			




	
	
	
		
 
    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],

			




	
	
	
		
 
     '%s://domain.org' % proto),

			




	
	
	
		
 
    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',

			




	
	
	
		
 
                                                '8080'],

			




	
	
	
		
 
     '%s://domain.org:8080' % proto),

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 
proto = 'https'

			




	
	
	
		
 
TEST_URLS += [

			




	
	
	
		
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

			




	
	
	
		
 
     '%s://127.0.0.1:8080' % proto),

			




	
	
	
		
 
    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],

			




	
	
	
		
 
     '%s://domain.org' % proto),

			




	
	
	
		
 
    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',

			




	
	
	
		
 
                                                '8080'],

			




	
	
	
		
 
     '%s://domain.org:8080' % proto),

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestLibs(BaseTestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand(TEST_URLS)

			




	
	
	
		
 
    def test_uri_filter(self, test_url, expected, expected_creds):

			




	
	
	
		
 
        from kallithea.lib.utils2 import uri_filter

			




	
	
	
		
 
        self.assertEqual(uri_filter(test_url), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand(TEST_URLS)

			




	
	
	
		
 
    def test_credentials_filter(self, test_url, expected, expected_creds):

			




	
	
	
		
 
        from kallithea.lib.utils2 import credentials_filter

			




	
	
	
		
 
        self.assertEqual(credentials_filter(test_url), expected_creds)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([('t', True),

			




	
	
	
		
 
                           ('true', True),

			




	
	
	
		
 
                           ('y', True),

			




	
	
	
		
 
                           ('yes', True),

			




	
	
	
		
 
                           ('on', True),

			




	
	
	
		
 
                           ('1', True),

			




	
	
	
		
 
                           ('Y', True),

			




	
	
	
		
 
                           ('yeS', True),

			




	
	
	
		
 
                           ('Y', True),

			




	
	
	
		
 
                           ('TRUE', True),

			




	
	
	
		
 
                           ('T', True),

			




	
	
	
		
 
                           ('False', False),

			




	
	
	
		
 
                           ('F', False),

			




	
	
	
		
 
                           ('FALSE', False),

			




	
	
	
		
 
                           ('0', False),

			




	
	
	
		
 
                           ('-1', False),

			




	
	
	
		
 
                           ('', False)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_str2bool(self, str_bool, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import str2bool

			




	
	
	
		
 
        self.assertEqual(str2bool(str_bool), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_mention_extractor(self):

			




	
	
	
		
 
        from kallithea.lib.utils2 import extract_mentioned_users

			




	
	
	
		
 
        sample = (

			




	
	
	
		
 
            "@first hi there @marcink here's my email marcin@email.com "

			




	
	
	
		
 
            "@first hi there @world here's my email username@email.com "

			




	
	
	
		
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "

			




	
	
	
		
 
            "@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl "

			




	
	
	
		
 
            "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "

			




	
	
	
		
 
            "@marian.user just do it @marco-polo and next extract @marco_polo "

			




	
	
	
		
 
            "user.dot  hej ! not-needed maril@domain.org"

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 
        s = sorted([

			




	
	
	
		
 
            '2one_more22', 'first', 'marcink', 'lukaszb', 'one', 'one_more22', 'MARCIN', 'maRCiN', 'john',

			




	
	
	
		
 
            'marian.user', 'marco-polo', 'marco_polo'], key=lambda k: k.lower())

			




	
	
	
		
 
            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',

			




	
	
	
		
 
            'marian.user', 'marco-polo', 'marco_polo', 'world'], key=lambda k: k.lower())

			




	
	
	
		
 
        self.assertEqual(s, extract_mentioned_users(sample))

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds= -1), u'1 second ago'),

			




	
	
	
		
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),

			




	
	
	
		
 
        (dict(hours= -1), u'1 hour ago'),

			




	
	
	
		
 
        (dict(hours= -24), u'1 day ago'),

			




	
	
	
		
 
        (dict(hours= -24 * 5), u'5 days ago'),

			




	
	
	
		
 
        (dict(months= -1), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -2), u'1 month and 2 days ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -20), u'1 month and 19 days ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -1), u'1 year and 1 month ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -10), u'1 year and 10 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -4), u'2 years and 4 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -11), u'2 years and 11 months ago'),

			




	
	
	
		
 
        (dict(years= -3, months= -2), u'3 years and 2 months ago'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds= -1), u'1 second ago'),

			




	
	
	
		
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),

			




	
	
	
		
 
        (dict(hours= -1), u'1 hour ago'),

			




	
	
	
		
 
        (dict(hours= -24), u'1 day ago'),

			




	
	
	
		
 
        (dict(hours= -24 * 5), u'5 days ago'),

			




	
	
	
		
 
        (dict(months= -1), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -2), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -20), u'1 month ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -1), u'13 months ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -10), u'22 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -4), u'2 years ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -11), u'3 years ago'),

			




	
	
	
		
 
        (dict(years= -3, months= -2), u'3 years ago'),

			




	
	
	
		
 
        (dict(years= -4, months= -8), u'5 years ago'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age_short(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), show_short_version=True, now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds=1), u'in 1 second'),

			




	
	
	
		
 
        (dict(seconds=60 * 2), u'in 2 minutes'),

			




	
	
	
		
 
        (dict(hours=1), u'in 1 hour'),

			




	
	
	
		
 
        (dict(hours=24), u'in 1 day'),

			




	
	
	
		
 
        (dict(hours=24 * 5), u'in 5 days'),

			




	
	
	
		
 
        (dict(months=1), u'in 1 month'),

			




	
	
	
		
 
        (dict(months=1, days=1), u'in 1 month and 1 day'),

			




	
	
	
		
 
        (dict(years=1, months=1), u'in 1 year and 1 month')

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age_in_future(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_tag_exctrator(self):

			




	
	
	
		
 
        sample = (

			




	
	
	
		
 
            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"

			




	
	
	
		
 
            "[requires] [stale] [see<>=>] [see => http://url.com]"

			




	
	
	
		
 
            "[requires => url] [lang => python] [just a tag]"

			




	
	
	
		
 
            "[,d] [ => ULR ] [obsolete] [desc]]"

			




	
	
	
		
 
        )

			




	
	
	
		
 
        from kallithea.lib.helpers import desc_stylize

			




	
	
	
		
 
        res = desc_stylize(sample)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="obsolete">obsolete</div>' in res)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="stale">stale</div>' in res)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="lang">python</div>' in res)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res)

			




	
	
	
		
 
        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_alternative_gravatar(self):

			




	
	
	
		
 
        from kallithea.lib.helpers import gravatar_url

			




	
	
	
		
 
        _md5 = lambda s: hashlib.md5(s).hexdigest()

			




	
	
	
		
 

			




	
	
	
		
 
        #mock pylons.url

			




	
	
	
		
 
        class fake_url(object):

			




	
	
	
		
 
            @classmethod

			




	
	
	
		
 
            def current(cls, *args, **kwargs):

			




	
	
	
		
 
                return 'https://server.com'

			




	
	
	
		
 

			




	
	
	
		
 
        #mock pylons.tmpl_context

			




	
	
	
		
 
        def fake_tmpl_context(_url):

			




	
	
	
		
 
            _c = AttributeDict()

			




	
	
	
		
 
            _c.visual = AttributeDict()

			




	
	
	
		
 
            _c.visual.use_gravatar = True

			




	
	
	
		
 
            _c.visual.gravatar_url = _url

			




	
	
	
		
 

			




	
	
	
		
 
            return _c

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
        with mock.patch('pylons.url', fake_url):

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://test.com/{email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                    from pylons import url

			




	
	
	
		
 
                    assert url.current() == 'https://server.com'

			




	
	
	
		
 
                    grav = gravatar_url(email_address='test@foo.com', size=24)

			




	
	
	
		
 
                    assert grav == 'http://test.com/test@foo.com'

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://test.com/{email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                grav = gravatar_url(email_address='test@foo.com', size=24)

			




	
	
	
		
 
                assert grav == 'http://test.com/test@foo.com'

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://test.com/{md5email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@foo.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'http://test.com/%s' % (_md5(em))

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://test.com/{md5email}/{size}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@foo.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@foo.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/rc', 'http://vps1:8000/rc/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc', 'http://user@vps1:8000/rc/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc', 'http://marcink@vps1:8000/rc/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc/', 'http://user@vps1:8000/rc/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc/', 'http://marcink@vps1:8000/rc/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),

			




	
	
	
		
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://vps1:8000/_23'),

			




	
	
	
		
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://marcink@proxy1.server.com/group/repo1'),

			




	
	
	
		
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),

			




	
	
	
		
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.server.com/group/repo1'),

			




	
	
	
		
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.server.com/group/repo1'),

			




	
	
	
		
 
        ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://proxy1.server.com/marcink/group/repo1'),

			




	
	
	
		
 
        ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.server.com/username/group/repo1'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import get_clone_url

			




	
	
	
		
 
        clone_url = get_clone_url(uri_tmpl=tmpl, qualifed_home_url='http://vps1:8000'+prefix,

			




	
	
	
		
 
                                  repo_name=repo_name, repo_id=23, **overrides)

			




	
	
	
		
 
        self.assertEqual(clone_url, expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Changes `some text url[foo]` => `some text <a href="/">foo</a>

			




	
	
	
		
 

			




	
	
	
		
 
        :param text:

			




	
	
	
		
 
        """

			




	
	
	
		
 
        import re

			




	
	
	
		
 
        # quickly change expected url[] into a link

			




	
	
	
		
 
        URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')

			




	
	
	
		
 

			




	
	
	
		
 
        def url_func(match_obj):

			




	
	
	
		
 
            _url = match_obj.groups()[0]

			




	
	
	
		
 
            return tmpl % (url_ or '/some-url', _url)

			




	
	
	
		
 
        return URL_PAT.sub(url_func, text)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
      ("",

			




	
	
	
		
 
       ""),

			




	
	
	
		
 
      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",

			




	
	
	
		
 
       "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),

			




	
	
	
		
 
      ("from rev 000000000000",

			




	
	
	
		
 
       "from rev url[000000000000]"),

			




	
	
	
		
 
      ("from rev 000000000000123123 also rev 000000000000",

			




	
	
	
		
 
       "from rev url[000000000000123123] also rev url[000000000000]"),

			




	
	
	
		
 
      ("this should-000 00",

			




	
	
	
		
 
       "this should-000 00"),

			




	
	
	
		
 
      ("longtextffffffffff rev 123123123123",

			




	
	
	
		
 
       "longtextffffffffff rev url[123123123123]"),

			




	
	
	
		
 
      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",

			




	
	
	
		
 
       "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),

			




	
	
	
		
 
      ("ffffffffffff some text traalaa",

			




	
	
	
		
 
       "url[ffffffffffff] some text traalaa"),

			




	
	
	
		
 
       ("""Multi line

			




	
	
	
		
 
       123123123123

			




	
	
	
		
 
       some text 123123123123

			




	
	
	
		
 
       sometimes !

			




	
	
	
		
 
       """,

			




	
	
	
		
 
       """Multi line

			




	
	
	
		
 
       url[123123123123]

			




	
	
	
		
 
       some text url[123123123123]

			




	
	
	
		
 
       sometimes !

			




	
	
	
		
 
       """)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_urlify_changesets(self, sample, expected):

			




	
	
	
		
 
        def fake_url(self, *args, **kwargs):

			




	
	
	
		
 
            return '/some-url'

			




	
	
	
		
 

			




	
	
	
		
 
        expected = self._quick_url(expected)

			




	
	
	
		
 

			




	
	
	
		
 
        with mock.patch('pylons.url', fake_url):

			




	
	
	
		
 
            from kallithea.lib.helpers import urlify_changesets

			




	
	
	
		
 
            self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
      ("",

			




	
	
	
		
 
       "",

			




	
	
	
		
 
       ""),

			




	
	
	
		
 
      ("https://svn.apache.org/repos",

			




	
	
	
		
 
       "url[https://svn.apache.org/repos]",

			




	
	
	
		
 
       "https://svn.apache.org/repos"),

			




	
	
	
		
 
      ("http://svn.apache.org/repos",

			




	
	
	
		
 
       "url[http://svn.apache.org/repos]",

			




	
	
	
		
 
       "http://svn.apache.org/repos"),

			




	
	
	
		
 
      ("from rev a also rev http://google.com",

			




	
	
	
		
 
       "from rev a also rev url[http://google.com]",

			




	
	
	
		
 
       "http://google.com"),

			




	
	
	
		
 
       ("""Multi line

			




	
	
	
		
 
       https://foo.bar.com

			




	
	
	
		
 
       some text lalala""",

			




	
	
	
		
 
       """Multi line

			




	
	
	
		
 
       url[https://foo.bar.com]

			




	
	
	
		
 
       some text lalala""",

			




	
	
	
		
 
       "https://foo.bar.com")

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_urlify_test(self, sample, expected, url_):

			




	
	
	
		
 
        from kallithea.lib.helpers import urlify_text

			




	
	
	
		
 
        expected = self._quick_url(expected,

			




	
	
	
		
 
                                   tmpl="""<a href="%s">%s</a>""", url_=url_)

			




	
	
	
		
 
        self.assertEqual(urlify_text(sample), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
      ("", None),

			




	
	
	
		
 
      ("/_2", '2'),

			




	
	
	
		
 
      ("_2", '2'),

			




	
	
	
		
 
      ("/_2/", '2'),

			




	
	
	
		
 
      ("_2/", '2'),

			




	
	
	
		
 

			




	
	
	
		
 
      ("/_21", '21'),

			




	
	
	
		
 
      ("_21", '21'),

			




	
	
	
		
 
      ("/_21/", '21'),

			




	
	
	
		
 
      ("_21/", '21'),

			




	
	
	
		
 

			




	
	
	
		
 
      ("/_21/foobar", '21'),

			




	
	
	
		
 
      ("_21/121", '21'),

			




	
	
	
		
 
      ("/_21/_12", '21'),

			




	
	
	
		
 
      ("_21/rc/foo", '21'),

			




	
	
	
		
 
      ("_21/prefix/foo", '21'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_get_repo_by_id(self, test, expected):

			




	
	
	
		
 
        from kallithea.lib.utils import _extract_id_from_repo_name

			




	
	
	
		
 
        _test = _extract_id_from_repo_name(test)

			




	
	
	
		
 
        self.assertEqual(_test, expected, msg='url:%s, got:`%s` expected: `%s`'

			




	
	
	
		
 
                                              % (test, _test, expected))

			




        

    


    
    

    

        

                

                    kallithea/tests/scripts/create_rc.sh
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
#!/bin/sh

			




	
	
	
		
 
psql -U postgres -h localhost -c 'drop database if exists kallithea;'

			




	
	
	
		
 
psql -U postgres -h localhost -c 'create database kallithea;'

			




	
	
	
		
 
paster setup-db rc.ini --force-yes --user=username --password=qweqwe --email=username@example.com --repos=/home/username/repos --no-public-access

			




	
	
	
		
 
paster setup-db server.ini --force-yes --user=username --password=qweqwe --email=username@example.com --repos=/home/username/repos --no-public-access

			




	
	
	
		
 
API_KEY=`psql -R " " -A -U postgres -h localhost -c "select api_key from users where admin=TRUE" -d kallithea | awk '{print $2}'`

			




	
	
	
		
 
echo "run those after running server"

			




	
	
	
		
 
paster serve rc.ini --pid-file=rc.pid --daemon

			




	
	
	
		
 
paster serve server.ini --pid-file=server.pid --daemon

			




	
	
	
		
 
sleep 3

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo1 password:qweqwe email:demo1@example.com

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo2 password:qweqwe email:demo2@example.com

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo3 password:qweqwe email:demo3@example.com

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user_group group_name:demo12

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo1

			




	
	
	
		
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo2

			




	
	
	
		
 
echo "killing server"

			




	
	
	
		
 
kill `cat rc.pid`

			




	
	
	
		
 
rm rc.pid

			




	
	
	
		
 
kill `cat server.pid`

			




	
	
	
		
 
rm server.pid

			




        

    


    
    

    

        

                

                    kallithea/tests/scripts/test_concurency.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 
# This program is free software: you can redistribute it and/or modify

			




	
	
	
		
 
# it under the terms of the GNU General Public License as published by

			




	
	
	
		
 
# the Free Software Foundation, either version 3 of the License, or

			




	
	
	
		
 
# (at your option) any later version.

			




	
	
	
		
 
#

			




	
	
	
		
 
# This program is distributed in the hope that it will be useful,

			




	
	
	
		
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of

			




	
	
	
		
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

			




	
	
	
		
 
# GNU General Public License for more details.

			




	
	
	
		
 
#

			




	
	
	
		
 
# You should have received a copy of the GNU General Public License

			




	
	
	
		
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

			




	
	
	
		
 
"""

			




	
	
	
		
 
kallithea.tests.test_hg_operations

			




	
	
	
		
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

			




	
	
	
		
 

			




	
	
	
		
 
Test suite for making push/pull operations

			




	
	
	
		
 

			




	
	
	
		
 
This file was forked by the Kallithea project in July 2014.

			




	
	
	
		
 
Original author and date, and relevant copyright and licensing information is below:

			




	
	
	
		
 
:created_on: Dec 30, 2010

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import os

			




	
	
	
		
 
import sys

			




	
	
	
		
 
import shutil

			




	
	
	
		
 
import logging

			




	
	
	
		
 
from os.path import join as jn

			




	
	
	
		
 
from os.path import dirname as dn

			




	
	
	
		
 

			




	
	
	
		
 
from tempfile import _RandomNameSequence

			




	
	
	
		
 
from subprocess import Popen, PIPE

			




	
	
	
		
 

			




	
	
	
		
 
from paste.deploy import appconfig

			




	
	
	
		
 
from sqlalchemy import engine_from_config

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.lib.utils import add_cache

			




	
	
	
		
 
from kallithea.model import init_model

			




	
	
	
		
 
from kallithea.model import meta

			




	
	
	
		
 
from kallithea.model.db import User, Repository

			




	
	
	
		
 
from kallithea.lib.auth import get_crypt_password

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests import TESTS_TMP_PATH, HG_REPO

			




	
	
	
		
 
from kallithea.config.environment import load_environment

			




	
	
	
		
 

			




	
	
	
		
 
rel_path = dn(dn(dn(dn(os.path.abspath(__file__)))))

			




	
	
	
		
 
conf = appconfig('config:rc.ini', relative_to=rel_path)

			




	
	
	
		
 
conf = appconfig('config:development.ini', relative_to=rel_path)

			




	
	
	
		
 
load_environment(conf.global_conf, conf.local_conf)

			




	
	
	
		
 

			




	
	
	
		
 
add_cache(conf)

			




	
	
	
		
 

			




	
	
	
		
 
USER = 'test_admin'

			




	
	
	
		
 
PASS = 'test12'

			




	
	
	
		
 
HOST = 'rc.local'

			




	
	
	
		
 
HOST = 'server.local'

			




	
	
	
		
 
METHOD = 'pull'

			




	
	
	
		
 
DEBUG = True

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class Command(object):

			




	
	
	
		
 

			




	
	
	
		
 
    def __init__(self, cwd):

			




	
	
	
		
 
        self.cwd = cwd

			




	
	
	
		
 

			




	
	
	
		
 
    def execute(self, cmd, *args):

			




	
	
	
		
 
        """Runs command on the system with given ``args``.

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        command = cmd + ' ' + ' '.join(args)

			




	
	
	
		
 
        log.debug('Executing %s' % command)

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print command

			




	
	
	
		
 
        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)

			




	
	
	
		
 
        stdout, stderr = p.communicate()

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print stdout, stderr

			




	
	
	
		
 
        return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def get_session():

			




	
	
	
		
 
    engine = engine_from_config(conf, 'sqlalchemy.db1.')

			




	
	
	
		
 
    init_model(engine)

			




	
	
	
		
 
    sa = meta.Session

			




	
	
	
		
 
    return sa

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def create_test_user(force=True):

			




	
	
	
		
 
    print 'creating test user'

			




	
	
	
		
 
    sa = get_session()

			




	
	
	
		
 

			




	
	
	
		
 
    user = sa.query(User).filter(User.username == USER).scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    if force and user is not None:

			




	
	
	
		
 
        print 'removing current user'

			




	
	
	
		
 
        for repo in sa.query(Repository).filter(Repository.user == user).all():

			




	
	
	
		
 
            sa.delete(repo)

			




	
	
	
		
 
        sa.delete(user)

			




	
	
	
		
 
        sa.commit()

			




	
	
	
		
 

			




	
	
	
		
 
    if user is None or force:

			




	
	
	
		
 
        print 'creating new one'

			




	
	
	
		
 
        new_usr = User()

			




	
	
	
		
 
        new_usr.username = USER

			




	
	
	
		
 
        new_usr.password = get_crypt_password(PASS)

			




	
	
	
		
 
        new_usr.email = 'mail@mail.com'

			




	
	
	
		
 
        new_usr.name = 'test'

			




	
	
	
		
 
        new_usr.lastname = 'lasttestname'

			




	
	
	
		
 
        new_usr.active = True

			




	
	
	
		
 
        new_usr.admin = True

			




	
	
	
		
 
        sa.add(new_usr)

			




	
	
	
		
 
        sa.commit()

			




	
	
	
		
 

			




	
	
	
		
 
    print 'done'

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def create_test_repo(force=True):

			




	
	
	
		
 
    print 'creating test repo'

			




	
	
	
		
 
    from kallithea.model.repo import RepoModel

			




	
	
	
		
 
    sa = get_session()

			




	
	
	
		
 

			




	
	
	
		
 
    user = sa.query(User).filter(User.username == USER).scalar()

			




	
	
	
		
 
    if user is None:

			




	
	
	
		
 
        raise Exception('user not found')

			




	
	
	
		
 

			




	
	
	
		
 
    repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    if repo is None:

			




	
	
	
		
 
        print 'repo not found creating'

			




	
	
	
		
 

			




	
	
	
		
 
        form_data = {'repo_name': HG_REPO,

			




	
	
	
		
 
                     'repo_type': 'hg',

			




	
	
	
		
 
                     'private':False,

			




	
	
	
		
 
                     'clone_uri': '' }

			




	
	
	
		
 
        rm = RepoModel(sa)

			




	
	
	
		
 
        rm.base_path = '/home/hg'

			




	
	
	
		
 
        rm.create(form_data, user)

			




	
	
	
		
 

			




	
	
	
		
 
    print 'done'

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def set_anonymous_access(enable=True):

			




	
	
	
		
 
    sa = get_session()

			




	
	
	
		
 
    user = sa.query(User).filter(User.username == 'default').one()

			




	
	
	
		
 
    user.active = enable

			




	
	
	
		
 
    sa.add(user)

			




	
	
	
		
 
    sa.commit()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def get_anonymous_access():

			




	
	
	
		
 
    sa = get_session()

			




        

    


    
    

    

        

                

                    kallithea/tests/scripts/test_crawler.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 
# This program is free software: you can redistribute it and/or modify

			




	
	
	
		
 
# it under the terms of the GNU General Public License as published by

			




	
	
	
		
 
# the Free Software Foundation, either version 3 of the License, or

			




	
	
	
		
 
# (at your option) any later version.

			




	
	
	
		
 
#

			




	
	
	
		
 
# This program is distributed in the hope that it will be useful,

			




	
	
	
		
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of

			




	
	
	
		
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

			




	
	
	
		
 
# GNU General Public License for more details.

			




	
	
	
		
 
#

			




	
	
	
		
 
# You should have received a copy of the GNU General Public License

			




	
	
	
		
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

			




	
	
	
		
 
"""

			




	
	
	
		
 
kallithea.tests.test_crawer

			




	
	
	
		
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~

			




	
	
	
		
 

			




	
	
	
		
 
Test for crawling a project for memory usage

			




	
	
	
		
 
This should be runned just as regular script together

			




	
	
	
		
 
with a watch script that will show memory usage.

			




	
	
	
		
 

			




	
	
	
		
 
watch -n1 ./kallithea/tests/mem_watch

			




	
	
	
		
 

			




	
	
	
		
 
This file was forked by the Kallithea project in July 2014.

			




	
	
	
		
 
Original author and date, and relevant copyright and licensing information is below:

			




	
	
	
		
 
:created_on: Apr 21, 2010

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
import cookielib

			




	
	
	
		
 
import urllib

			




	
	
	
		
 
import urllib2

			




	
	
	
		
 
import time

			




	
	
	
		
 
import os

			




	
	
	
		
 
import sys

			




	
	
	
		
 
from os.path import join as jn

			




	
	
	
		
 
from os.path import dirname as dn

			




	
	
	
		
 

			




	
	
	
		
 
__here__ = os.path.abspath(__file__)

			




	
	
	
		
 
__root__ = dn(dn(dn(__here__)))

			




	
	
	
		
 
sys.path.append(__root__)

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.lib import vcs

			




	
	
	
		
 
from kallithea.lib.compat import OrderedSet

			




	
	
	
		
 
from kallithea.lib.vcs.exceptions import RepositoryError

			




	
	
	
		
 

			




	
	
	
		
 
PASES = 3

			




	
	
	
		
 
HOST = 'http://127.0.0.1'

			




	
	
	
		
 
PORT = 5000

			




	
	
	
		
 
BASE_URI = '%s:%s/' % (HOST, PORT)

			




	
	
	
		
 

			




	
	
	
		
 
if len(sys.argv) == 2:

			




	
	
	
		
 
    BASE_URI = sys.argv[1]

			




	
	
	
		
 

			




	
	
	
		
 
if not BASE_URI.endswith('/'):

			




	
	
	
		
 
    BASE_URI += '/'

			




	
	
	
		
 

			




	
	
	
		
 
print 'Crawling @ %s' % BASE_URI

			




	
	
	
		
 
BASE_URI += '%s'

			




	
	
	
		
 
PROJECT_PATH = jn('/', 'home', 'marcink', 'repos')

			




	
	
	
		
 
PROJECT_PATH = jn('/', 'home', 'username', 'repos')

			




	
	
	
		
 
PROJECTS = [

			




	
	
	
		
 
    #'linux-magx-pbranch',

			




	
	
	
		
 
    'CPython',

			




	
	
	
		
 
    'kallithea',

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
cj = cookielib.FileCookieJar('/tmp/rc_test_cookie.txt')

			




	
	
	
		
 
o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))

			




	
	
	
		
 
o.addheaders = [

			




	
	
	
		
 
    ('User-agent', 'kallithea-crawler'),

			




	
	
	
		
 
    ('Accept-Language', 'en - us, en;q = 0.5')

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 
urllib2.install_opener(o)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_repo(proj):

			




	
	
	
		
 
    if isinstance(proj, basestring):

			




	
	
	
		
 
        repo = vcs.get_repo(jn(PROJECT_PATH, proj))

			




	
	
	
		
 
        proj = proj

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        repo = proj

			




	
	
	
		
 
        proj = repo.name

			




	
	
	
		
 

			




	
	
	
		
 
    return repo, proj

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def test_changelog_walk(proj, pages=100):

			




	
	
	
		
 
    repo, proj = _get_repo(proj)

			




	
	
	
		
 

			




	
	
	
		
 
    total_time = 0

			




	
	
	
		
 
    for i in range(1, pages):

			




	
	
	
		
 

			




	
	
	
		
 
        page = '/'.join((proj, 'changelog',))

			




	
	
	
		
 

			




	
	
	
		
 
        full_uri = (BASE_URI % page) + '?' + urllib.urlencode({'page': i})

			




	
	
	
		
 
        s = time.time()

			




	
	
	
		
 
        f = o.open(full_uri)

			




	
	
	
		
 

			




	
	
	
		
 
        assert f.url == full_uri, 'URL:%s does not match %s' % (f.url, full_uri)

			




	
	
	
		
 

			




	
	
	
		
 
        size = len(f.read())

			




	
	
	
		
 
        e = time.time() - s

			




	
	
	
		
 
        total_time += e

			




	
	
	
		
 
        print 'visited %s size:%s req:%s ms' % (full_uri, size, e)

			




	
	
	
		
 

			




	
	
	
		
 
    print 'total_time', total_time

			




	
	
	
		
 
    print 'average on req', total_time / float(pages)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def test_changeset_walk(proj, limit=None):

			




	
	
	
		
 
    repo, proj = _get_repo(proj)

			




	
	
	
		
 

			




	
	
	
		
 
    print 'processing', jn(PROJECT_PATH, proj)

			




	
	
	
		
 
    total_time = 0

			




	
	
	
		
 

			




	
	
	
		
 
    cnt = 0

			




	
	
	
		
 
    for i in repo:

			




	
	
	
		
 
        cnt += 1

			




	
	
	
		
 
        raw_cs = '/'.join((proj, 'changeset', i.raw_id))

			




	
	
	
		
 
        if limit and limit == cnt:

			




	
	
	
		
 
            break

			




	
	
	
		
 

			




	
	
	
		
 
        full_uri = (BASE_URI % raw_cs)

			




	
	
	
		
 
        print '%s visiting %s\%s' % (cnt, full_uri, i)

			




	
	
	
		
 
        s = time.time()

			




	
	
	
		
 
        f = o.open(full_uri)

			




	
	
	
		
 
        size = len(f.read())

			




	
	
	
		
 
        e = time.time() - s

			




	
	
	
		
 
        total_time += e

			




	
	
	
		
 
        print '%s visited %s\%s size:%s req:%s ms' % (cnt, full_uri, i, size, e)

			




	
	
	
		
 

			




	
	
	
		
 
    print 'total_time', total_time

			




	
	
	
		
 
    print 'average on req', total_time / float(cnt)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def test_files_walk(proj, limit=100):

			




	
	
	
		
 
    repo, proj = _get_repo(proj)

			




	
	
	
		
 

			




	
	
	
		
 
    print 'processing', jn(PROJECT_PATH, proj)

			




	
	
	
		
 
    total_time = 0

			




	
	
	
		
 

			




	
	
	
		
 
    paths_ = OrderedSet([''])

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        tip = repo.get_changeset('tip')

			




	
	
	
		
 
        for topnode, dirs, files in tip.walk('/'):

			




	
	
	
		
 

			




	
	
	
		
 
            for dir in dirs:

			




	
	
	
		
 
                paths_.add(dir.path)

			




	
	
	
		
 
                for f in dir:

			




	
	
	
		
 
                    paths_.add(f.path)

			




	
	
	
		
 

			




	
	
	
		
 
            for f in files:

			




	
	
	
		
 
                paths_.add(f.path)

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/vcs/test_utils.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -92,188 +92,188 @@ class TestParseChangesets(unittest.TestC

			




	
	
	
		
 
            'end': None,

			




	
	
	
		
 
        })

			




	
	
	
		
 

			




	
	
	
		
 
    def test_end_is_returned_correctly(self):

			




	
	
	
		
 
        self.assertEqual(parse_changesets('..cccddd'), {

			




	
	
	
		
 
            'start': None,

			




	
	
	
		
 
            'main': None,

			




	
	
	
		
 
            'end': 'cccddd',

			




	
	
	
		
 
        })

			




	
	
	
		
 

			




	
	
	
		
 
    def test_that_two_or_three_dots_are_allowed(self):

			




	
	
	
		
 
        text1 = 'a..b'

			




	
	
	
		
 
        text2 = 'a...b'

			




	
	
	
		
 
        self.assertEqual(parse_changesets(text1), parse_changesets(text2))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_that_input_is_stripped_first(self):

			




	
	
	
		
 
        text1 = 'a..bb'

			




	
	
	
		
 
        text2 = '  a..bb\t\n\t '

			




	
	
	
		
 
        self.assertEqual(parse_changesets(text1), parse_changesets(text2))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_that_exception_is_raised(self):

			




	
	
	
		
 
        text = '123456.789012' # single dot is not recognized

			




	
	
	
		
 
        with self.assertRaises(ValueError):

			




	
	
	
		
 
            parse_changesets(text)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_non_alphanumeric_raises_exception(self):

			




	
	
	
		
 
        with self.assertRaises(ValueError):

			




	
	
	
		
 
            parse_changesets('aaa@bbb')

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestParseDatetime(unittest.TestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_datetime_text(self):

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2010-04-07 21:29:41'),

			




	
	
	
		
 
            datetime.datetime(2010, 4, 7, 21, 29, 41))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_no_seconds(self):

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2010-04-07 21:29'),

			




	
	
	
		
 
            datetime.datetime(2010, 4, 7, 21, 29))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_date_only(self):

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2010-04-07'),

			




	
	
	
		
 
            datetime.datetime(2010, 4, 7))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_another_format(self):

			




	
	
	
		
 
        self.assertEqual(parse_datetime('04/07/10 21:29:41'),

			




	
	
	
		
 
            datetime.datetime(2010, 4, 7, 21, 29, 41))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_now(self):

			




	
	
	
		
 
        self.assertTrue(parse_datetime('now') - datetime.datetime.now() <

			




	
	
	
		
 
            datetime.timedelta(seconds=1))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_today(self):

			




	
	
	
		
 
        today = datetime.date.today()

			




	
	
	
		
 
        self.assertEqual(parse_datetime('today'),

			




	
	
	
		
 
            datetime.datetime(*today.timetuple()[:3]))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_yesterday(self):

			




	
	
	
		
 
        yesterday = datetime.date.today() - datetime.timedelta(days=1)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('yesterday'),

			




	
	
	
		
 
            datetime.datetime(*yesterday.timetuple()[:3]))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_tomorrow(self):

			




	
	
	
		
 
        tomorrow = datetime.date.today() + datetime.timedelta(days=1)

			




	
	
	
		
 
        args = tomorrow.timetuple()[:3] + (23, 59, 59)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('tomorrow'), datetime.datetime(*args))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_days(self):

			




	
	
	
		
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3)

			




	
	
	
		
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)

			




	
	
	
		
 
        expected = datetime.datetime(*args)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3d'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 d'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 day'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 days'), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_weeks(self):

			




	
	
	
		
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3 * 7)

			




	
	
	
		
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)

			




	
	
	
		
 
        expected = datetime.datetime(*args)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3w'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 w'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 week'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('3 weeks'), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_mixed(self):

			




	
	
	
		
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=2 * 7 + 3)

			




	
	
	
		
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)

			




	
	
	
		
 
        expected = datetime.datetime(*args)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2w3d'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2w 3d'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2w 3 days'), expected)

			




	
	
	
		
 
        self.assertEqual(parse_datetime('2 weeks 3 days'), expected)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestAuthorExtractors(unittest.TestCase):

			




	
	
	
		
 
    TEST_AUTHORS = [('Marcin Kuzminski <marcin@python-works.com>',

			




	
	
	
		
 
                    ('Marcin Kuzminski', 'marcin@python-works.com')),

			




	
	
	
		
 
                  ('Marcin Kuzminski Spaces < marcin@python-works.com >',

			




	
	
	
		
 
                    ('Marcin Kuzminski Spaces', 'marcin@python-works.com')),

			




	
	
	
		
 
                  ('Marcin Kuzminski <marcin.kuzminski@python-works.com>',

			




	
	
	
		
 
                    ('Marcin Kuzminski', 'marcin.kuzminski@python-works.com')),

			




	
	
	
		
 
                  ('mrf RFC_SPEC <marcin+kuzminski@python-works.com>',

			




	
	
	
		
 
                    ('mrf RFC_SPEC', 'marcin+kuzminski@python-works.com')),

			




	
	
	
		
 
    TEST_AUTHORS = [("Username Last'o'Name <username@python-works.com>",

			




	
	
	
		
 
                    ("Username Last'o'Name", "username@python-works.com")),

			




	
	
	
		
 
                  ("Username Last'o'Name Spaces < username@python-works.com >",

			




	
	
	
		
 
                    ("Username Last'o'Name Spaces", "username@python-works.com")),

			




	
	
	
		
 
                  ("Username Last'o'Name <username.lastname@python-works.com>",

			




	
	
	
		
 
                    ("Username Last'o'Name", "username.lastname@python-works.com")),

			




	
	
	
		
 
                  ('mrf RFC_SPEC <username+lastname@python-works.com>',

			




	
	
	
		
 
                    ('mrf RFC_SPEC', 'username+lastname@python-works.com')),

			




	
	
	
		
 
                  ('username <user@email.com>',

			




	
	
	
		
 
                    ('username', 'user@email.com')),

			




	
	
	
		
 
                  ('username <user@email.com',

			




	
	
	
		
 
                   ('username', 'user@email.com')),

			




	
	
	
		
 
                  ('broken missing@email.com',

			




	
	
	
		
 
                   ('broken', 'missing@email.com')),

			




	
	
	
		
 
                  ('<justemail@mail.com>',

			




	
	
	
		
 
                   ('', 'justemail@mail.com')),

			




	
	
	
		
 
                  ('justname',

			




	
	
	
		
 
                   ('justname', '')),

			




	
	
	
		
 
                  ('Mr Double Name withemail@email.com ',

			




	
	
	
		
 
                   ('Mr Double Name', 'withemail@email.com')),

			




	
	
	
		
 
                  ]

			




	
	
	
		
 

			




	
	
	
		
 
    def test_author_email(self):

			




	
	
	
		
 

			




	
	
	
		
 
        for test_str, result in self.TEST_AUTHORS:

			




	
	
	
		
 
            self.assertEqual(result[1], author_email(test_str))

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
    def test_author_name(self):

			




	
	
	
		
 

			




	
	
	
		
 
        for test_str, result in self.TEST_AUTHORS:

			




	
	
	
		
 
            self.assertEqual(result[0], author_name(test_str))

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestGetDictForAttrs(unittest.TestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_returned_dict_has_expected_attrs(self):

			




	
	
	
		
 
        obj = mock.Mock()

			




	
	
	
		
 
        obj.NOT_INCLUDED = 'this key/value should not be included'

			




	
	
	
		
 
        obj.CONST = True

			




	
	
	
		
 
        obj.foo = 'aaa'

			




	
	
	
		
 
        obj.attrs = {'foo': 'bar'}

			




	
	
	
		
 
        obj.date = datetime.datetime(2010, 12, 31)

			




	
	
	
		
 
        obj.count = 1001

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual(get_dict_for_attrs(obj, ['CONST', 'foo', 'attrs',

			




	
	
	
		
 
            'date', 'count']), {

			




	
	
	
		
 
            'CONST': True,

			




	
	
	
		
 
            'foo': 'aaa',

			




	
	
	
		
 
            'attrs': {'foo': 'bar'},

			




	
	
	
		
 
            'date': datetime.datetime(2010, 12, 31),

			




	
	
	
		
 
            'count': 1001,

			




	
	
	
		
 
        })

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestGetTotalSeconds(unittest.TestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    def assertTotalSecondsEqual(self, timedelta, expected_seconds):

			




	
	
	
		
 
        result = get_total_seconds(timedelta)

			




	
	
	
		
 
        self.assertEqual(result, expected_seconds,

			




	
	
	
		
 
            "We computed %s seconds for %s but expected %s"

			




	
	
	
		
 
            % (result, timedelta, expected_seconds))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_get_total_seconds_returns_proper_value(self):

			




	
	
	
		
 
        self.assertTotalSecondsEqual(datetime.timedelta(seconds=1001), 1001)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_get_total_seconds_returns_proper_value_for_partial_seconds(self):

			




	
	
	
		
 
        self.assertTotalSecondsEqual(datetime.timedelta(seconds=50.65), 50.65)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestGetUserHome(unittest.TestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    @mock.patch.object(os, 'environ', {})

			




	
	
	
		
 
    def test_defaults_to_none(self):

			




	
	
	
		
 
        self.assertEqual(get_user_home(), '')

			




	
	
	
		
 

			




	
	
	
		
 
    @mock.patch.object(os, 'environ', {'HOME': '/home/foobar'})

			




	
	
	
		
 
    def test_unix_like(self):

			




	
	
	
		
 
        self.assertEqual(get_user_home(), '/home/foobar')

			




	
	
	
		
 

			




	
	
	
		
 
    @mock.patch.object(os, 'environ', {'USERPROFILE': '/Users/foobar'})

			




	
	
	
		
 
    def test_windows_like(self):

			




	
	
	
		
 
        self.assertEqual(get_user_home(), '/Users/foobar')

			




	
	
	
		
 

			




	
	
	
		
 
    @mock.patch.object(os, 'environ', {'HOME': '/home/foobar',

			




	
	
	
		
 
        'USERPROFILE': '/Users/foobar'})

			




	
	
	
		
 
    def test_prefers_home_over_userprofile(self):

			




	
	
	
		
 
        self.assertEqual(get_user_home(), '/home/foobar')

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
if __name__ == '__main__':

			




	
	
	
		
 
    unittest.main()

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.