from kallithea.lib import webutils

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.base import BaseController, render

from kallithea.lib.utils import repo2db_mapper, set_app_settings

from kallithea.lib.utils2 import safe_str

from kallithea.lib.vcs import VCSError

from kallithea.lib.webutils import url

from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm

from kallithea.model.notification import EmailNotificationModel

from kallithea.model.scm import ScmModel

            recipients = [test_email] if test_email else None

                             test_email_txt_body, test_email_html_body)

            webutils.flash(_('Send email task created'), category='success')

        if request.POST:

            repo_location = self._get_hg_ui_settings()['paths_root_path']

            full_index = request.POST.get('full_index', False)

            webutils.flash(_('Whoosh reindex task scheduled'), category='success')

            raise HTTPFound(location=url('admin_settings_search'))

from kallithea.lib import ext_json, webutils

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController, jsonify, render

from kallithea.lib.conf import ALL_EXTS, ALL_READMES, LANGUAGES_EXTENSIONS_MAP

from kallithea.lib.markup_renderer import MarkupRenderer

from kallithea.lib.page import Page

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.lib.vcs.exceptions import ChangesetError, EmptyRepositoryError, NodeDoesNotExistError

from kallithea.lib.vcs.nodes import FileNode

log = logging.getLogger(__name__)

            c.trending_languages = []

        recurse_limit = 500  # don't recurse more than 500 times when parsing

        return render('summary/statistics.html')

class CeleryConfig(object):

    task_always_eager = False

# map from Kallithea .ini Celery 3 config names to Celery 4 config names

# 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 task modules, containing all task that suppose to be run

by celery daemon

from kallithea.lib.indexers.daemon import WhooshIndexingDaemon

from kallithea.lib.utils2 import asbool, ascii_bytes

from kallithea.lib.vcs.utils import author_email

__all__ = ['whoosh_index', 'get_commits_stats', 'send_email']

        co_day_auth_aggr = {}

        commits_by_day_aggregate = {}

            return True

        # return if repo have no revisions

        if repo_size < 1:

            lock.release()

        if cur_stats is not None:

            last_rev = cur_stats.stat_on_revision

            # pass silently without any work if we're not on first revision or

            # current state of parsing revision(from db marker) is the

            # last revision

        log.debug('Getting revisions from %s to %s',

             last_rev, last_rev + parse_limit

        )

            log.debug('parsing %s', cs)

            last_cs = cs  # remember last parsed changeset

            tt = cs.date.timetuple()

                               key=itemgetter(0))

        if not co_day_auth_aggr:

                "data": [0, 1],

                "schema": ["commits"],

            }

        stats.commit_activity_combined = ascii_bytes(ext_json.dumps(overview_data))

        log.debug('last revision %s', last_rev)

        log.debug('revisions to parse %s', leftovers)

        if last_rev == 0 or leftovers < parse_limit:

        lock.release()

        # execute another task if celery is enabled

            get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit - 1)

        elif recurse_limit <= 0:

            log.debug('Not recursing - limit has been reached')

    enable_downloads = defs.get('repo_enable_downloads')

    try:

            repo_name=repo_name_full,

            repo_type=repo_type,

            description=description,

        DBS.commit()

        # now create this repo on Filesystem

            repo_name=repo_name,

            repo_type=repo_type,

            repo_group=db.RepoGroup.guess_instance(repo_group),

            clone_uri=clone_uri,

        )

        # update repo changeset caches initially

        # set new created state

        DBS.commit()

    except Exception as e:

        log.warning('Exception %s occurred when forking repository, '

                    'doing cleanup...' % e)

        # rollback things manually !

            DBS.commit()

        raise

    return True

    try:

        fork_of = db.Repository.guess_instance(form_data.get('fork_parent_id'))

            repo_name=repo_name_full,

            repo_type=repo_type,

            description=form_data['description'],

        source_repo_path = os.path.join(base_path, fork_of.repo_name)

        # now create this repo on Filesystem

            repo_name=repo_name,

            repo_type=repo_type,

            repo_group=db.RepoGroup.guess_instance(repo_group),

            clone_uri=source_repo_path,

        )

        # update repo changeset caches initially

        # set new created state

        DBS.commit()

    except Exception as e:

        log.warning('Exception %s occurred when forking repository, '

                    'doing cleanup...' % e)

        # rollback things manually !

            DBS.commit()

        raise

    return True

def __get_codes_stats(repo_name):

    code_stats = {}

    for _topnode, _dirnodes, filenodes in tip.walk('/'):

from tg.i18n import ugettext as _

from kallithea.lib.utils2 import fmt_date

log = logging.getLogger(__name__)

        :param email_kwargs: additional dict to pass as args to email template

        """

        import kallithea.lib.helpers as h

        email_kwargs = email_kwargs or {}

        if recipients and not getattr(recipients, '__iter__', False):

            raise Exception('recipients must be a list or iterable')

        # send email with notification to participants

        for rec_mail in sorted(rec_mails):

                     email_html_body, headers,

                     from_name=created_by_obj.full_name_or_username)

        :param form_data:

        :param cur_user:

        """

    def _update_permissions(self, repo, perms_new=None, perms_updates=None,

                            check_perms=True):

        :param form_data:

        :param cur_user:

        """

    def delete(self, repo, forks=None, fs_remove=True, cur_user=None):

        """

        allowing users to copy-paste or manually enter the token from the

        email.

        """

        user_email = data['email']

        user = db.User.get_by_email(user_email)

                reset_token=token,

                reset_url=link)

            log.debug('sending email')

            log.info('send new password mail to %s', user_email)

        else:

            log.debug("password reset email %s not found", user_email)

        return expected_token == token

    def reset_password(self, user_email, new_passwd):

        user = db.User.get_by_email(user_email)

        if user is not None:

            if not self.can_change_password(user):

        if new_passwd is None:

            raise Exception('unable to set new password')

                 _('Password reset notification'),

                 _('The password to your account %s has been changed using password reset form.') % (user.username,))

        log.info('send password reset mail to %s', user_email)

import mock

from tg.util.webtest import test_context

from kallithea.lib import webutils

from kallithea.lib.utils2 import check_password, generate_api_key

from kallithea.model import db, meta, validators

        def mock_send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):

            collected.append((recipients, subject, body, html_body))

                mock.patch.object(time, 'time', lambda: timestamp):

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

                                         action='password_reset'),

from tg.util.webtest import test_context

import kallithea.lib.celerylib

import kallithea.lib.helpers as h

from kallithea.model import db, meta

from kallithea.model.notification import EmailNotificationModel, NotificationModel

from kallithea.model.user import UserModel

                assert body == "hi there"

                assert '>hi there<' in html_body

                assert from_name == 'u1 u1'

                NotificationModel().create(created_by=self.u1,

                                                   body='hi there',

                                                   recipients=usrs)

            l.append('<hr/>\n')

        with test_context(self.app):

                pr_kwargs = dict(

                    pr_nice_id='#7',

                    pr_title='The Title',

                # Email type TYPE_PASSWORD_RESET has no corresponding notification type - test it directly:

                desc = 'TYPE_PASSWORD_RESET'

                kwargs = dict(user='John Doe', reset_token='decbf64715098db5b0bd23eab44bd792670ab746', reset_url='http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746')

                    "Password reset link",

                    EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'txt', **kwargs),

                    EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'html', **kwargs),

        smtplib_mock.lastmsg = msg

class TestMail(base.TestController):

    def test_send_mail_trivial(self):

            'smtp_server': mailserver,

            'app_email_from': envelope_from,

        }

        assert smtplib_mock.lastdest == set(recipients)

        assert smtplib_mock.lastsender == envelope_from

            'app_email_from': envelope_from,

            'email_to': email_to,

        }

        assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL, email_to])

        assert smtplib_mock.lastsender == envelope_from

            'app_email_from': envelope_from,

            'email_to': email_to,

        }

        assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL] + email_to.split(','))

        assert smtplib_mock.lastsender == envelope_from

            'smtp_server': mailserver,

            'app_email_from': envelope_from,

        }

        assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL])

        assert smtplib_mock.lastsender == envelope_from

            'smtp_server': mailserver,

            'app_email_from': envelope_from,

        }

        assert smtplib_mock.lastdest == set(recipients)

        assert smtplib_mock.lastsender == envelope_from

            'smtp_server': mailserver,

            'app_email_from': envelope_from,

        }

        assert smtplib_mock.lastdest == set(recipients)

        assert smtplib_mock.lastsender == envelope_from

            'smtp_server': mailserver,

            'app_email_from': envelope_from,

        }

                                                     from_name=author.full_name_or_username, headers=headers)

        assert smtplib_mock.lastdest == set(recipients)

normal_modules = set('''

kallithea

kallithea.lib

kallithea.lib.auth

kallithea.lib.auth_modules

kallithea.lib.vcs

kallithea.lib.webutils

kallithea.model

kallithea.model.scm

kallithea.templates.py

'''.split())

('kallithea.lib.utils', 'kallithea.model'),  # clean up utils

('kallithea.lib.utils', 'kallithea.model.db'),

('kallithea.lib.utils', 'kallithea.model.scm'),

('kallithea.model', 'kallithea.lib.auth'),  # auth.HasXXX

('kallithea.model', 'kallithea.lib.auth_modules'),  # validators

('kallithea.model', 'kallithea.lib.helpers'),